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_with_serial_local_copy.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 32 : AlignedReduceScatterDoubleRingWithSerialLocalCopy::AlignedReduceScatterDoubleRingWithSerialLocalCopy(
16 32 : const HcclDispatcher dispatcher)
17 32 : : AlignedReduceScatterDoubleRing(dispatcher)
18 32 : {}
19 :
20 64 : AlignedReduceScatterDoubleRingWithSerialLocalCopy::~AlignedReduceScatterDoubleRingWithSerialLocalCopy() {}
21 :
22 : // reduce scatter ring direct算法的函数入口
23 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::RunAsync(
24 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
25 : {
26 : // 基本的检查
27 0 : CHK_RET(CheckParameters(rank, rankSize, links));
28 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy]RunAsync start");
29 :
30 : // 判断rank_size == 1的情况，并拷贝
31 0 : if (rankSize == 1) {
32 0 : CHK_RET(OneRankMemcpy());
33 0 : return HCCL_SUCCESS;
34 : }
35 : // 收集本地mem信息
36 0 : CHK_RET(InitSenderReducer());
37 :
38 : // 收集邻居信息
39 0 : CHK_RET(GetInitializedNeighborLinks(rank, rankSize, links));
40 :
41 : // 填充slice_
42 0 : CHK_RET(SetSlices(rank, rankSize));
43 :
44 : // 运行reduce-scatter, ring算法
45 0 : CHK_RET(RunReduceScatter(rank, rankSize));
46 :
47 0 : if (barrierSwitchOn_) {
48 : // 执行barrier，保证数据收发完成
49 0 : CHK_RET(ExecuteBarrier(leftLink_, rightLink_));
50 : }
51 :
52 0 : CHK_RET(LaunchTaskExtend(dispatcher_, stream_, subStreams_));
53 :
54 0 : HCCL_INFO("AlignedReduceScatterDoubleRingWithSerialLocalCopy finished: rank[%u] end", rank);
55 0 : return HCCL_SUCCESS;
56 : }
57 :
58 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::MemcpyInitSlices(
59 : u64 ringIndex, DeviceMem& dstInit, DeviceMem& srcInit, DeviceMem& dstSubInit, DeviceMem& srcSubInit)
60 : {
61 0 : if (ringIndex == 1) {
62 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, stream_));
63 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstSubInit, srcSubInit, stream_));
64 : } else {
65 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB && (!disableDMAReduce_)) {
66 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, subStreams_[0]));
67 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstSubInit, srcSubInit, subStreams_[1]));
68 : } else {
69 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, subStreams_[0]));
70 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstSubInit, srcSubInit, subStreams_[0]));
71 : }
72 : }
73 0 : return HCCL_SUCCESS;
74 : }
75 :
76 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::RunMainInitStep(const u32 rank, const u32 rankSize)
77 : {
78 : // 主环初始indexes
79 0 : u32 initSlice0Idx = (rankSize - rank - 1 + rankSize) % rankSize;
80 0 : u32 initSlice1Idx = (rankSize - rank - DMA_REDUCE_TWO_OFFSET + rankSize) % rankSize;
81 0 : u32 discontinuousSliceSize = multRingsSlices_[ALIGNED_MAIN_RING_INDEX].size() / rankSize;
82 0 : HCCL_DEBUG("Memcpy operation: step[-1] starts on ring[%u]", ALIGNED_MAIN_RING_INDEX);
83 0 : DeviceMem dstInit;
84 0 : DeviceMem srcInit;
85 0 : DeviceMem dstSubInit;
86 0 : DeviceMem srcSubInit;
87 0 : for (u32 discontinuousSliceIdx = 0; discontinuousSliceIdx < discontinuousSliceSize; discontinuousSliceIdx++) {
88 0 : CHK_RET(PrepareInitSlices(
89 : rankSize, ALIGNED_MAIN_RING_INDEX, discontinuousSliceSize, discontinuousSliceIdx, initSlice0Idx,
90 : initSlice1Idx, dstInit, srcInit, dstSubInit, srcSubInit));
91 0 : CHK_RET(MemcpyInitSlices(ALIGNED_MAIN_RING_INDEX, dstInit, srcInit, dstSubInit, srcSubInit));
92 : }
93 0 : return HCCL_SUCCESS;
94 0 : }
95 :
96 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::RunSubInitStep(const u32 rank, const u32 rankSize)
97 : {
98 : // 从环初始indexes
99 0 : u32 initSlice0Idx = (rank + rankSize - 1) % rankSize;
100 0 : u32 initSlice1Idx = (rank + rankSize - DMA_REDUCE_TWO_OFFSET) % rankSize;
101 0 : u32 discontinuousSliceSize = multRingsSlices_[ALIGNED_SUB_RING_INDEX].size() / rankSize;
102 0 : HCCL_DEBUG("Memcpy operation: step[-1] starts on ring[%u]", ALIGNED_SUB_RING_INDEX);
103 0 : DeviceMem dstInit;
104 0 : DeviceMem srcInit;
105 0 : DeviceMem dstSubInit;
106 0 : DeviceMem srcSubInit;
107 0 : for (u32 discontinuousSliceIdx = 0; discontinuousSliceIdx < discontinuousSliceSize; discontinuousSliceIdx++) {
108 0 : CHK_RET(PrepareInitSlices(
109 : rankSize, ALIGNED_SUB_RING_INDEX, discontinuousSliceSize, discontinuousSliceIdx, initSlice0Idx,
110 : initSlice1Idx, dstInit, srcInit, dstSubInit, srcSubInit));
111 0 : CHK_RET(MemcpyInitSlices(ALIGNED_SUB_RING_INDEX, dstInit, srcInit, dstSubInit, srcSubInit));
112 : }
113 0 : return HCCL_SUCCESS;
114 0 : }
115 :
116 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::LocalMemcpy(
117 : const u32 step, const u32 rankSize, const u32 ringIndex, DeviceMem& localSrcMem, DeviceMem& localDstMem)
118 : {
119 : // 先调通单算子模式
120 : // 通过校验流数判断是单算子模式还是图模式
121 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB && (!disableDMAReduce_)) {
122 0 : if (ringIndex == 0) {
123 0 : CHK_RET(LocalNotify::Post(
124 : subStreams_[ringIndex + 1], dispatcher_, mainSignals_[ringIndex + 1], profilerInput_.stage));
125 0 : CHK_RET(LocalNotify::Wait(
126 : subStreams_[ringIndex + 1], dispatcher_, subSignals_[ringIndex + 1], profilerInput_.stage));
127 0 : if (localSrcMem != localDstMem && step != rankSize - DMA_REDUCE_TWO_OFFSET) {
128 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, subStreams_[ringIndex + 1]));
129 : }
130 : } else {
131 0 : if (localSrcMem != localDstMem && step != rankSize - DMA_REDUCE_TWO_OFFSET) {
132 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, stream_));
133 : }
134 : }
135 : } else {
136 0 : if (ringIndex == 0) {
137 0 : if (localSrcMem != localDstMem && step != rankSize - DMA_REDUCE_TWO_OFFSET) {
138 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, subStreams_[0]));
139 : }
140 : } else {
141 0 : if (localSrcMem != localDstMem && step != rankSize - DMA_REDUCE_TWO_OFFSET) {
142 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localDstMem, localSrcMem, stream_));
143 : }
144 : }
145 : }
146 0 : return HCCL_SUCCESS;
147 : }
148 :
149 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::RunMainRingSubStream(const u32 rank, const u32 rankSize)
150 : {
151 : // 主环初始indexes
152 0 : u32 txSliceIdxMain = (rankSize - rank - 1 + rankSize) % rankSize;
153 0 : u32 rxSliceIdxMain = (rankSize - rank - DMA_REDUCE_TWO_OFFSET + rankSize) % rankSize;
154 0 : u32 subSliceIdxMain = (rankSize - rank - DMA_REDUCE_THREE_OFFSET + rankSize) % rankSize;
155 0 : for (u32 step = 0; step < rankSize - 1; step++) {
156 : // 并发
157 0 : std::vector<SenderMemoryInfo> txReduceMemsMain;
158 0 : std::vector<ReducerMemoryInfo> rxReduceMemsMain;
159 0 : std::vector<DeviceMem> localSrcMemsMain;
160 0 : std::vector<DeviceMem> localDstMemsMain;
161 0 : CHK_RET(PrepareDeviceMems(
162 : step, ALIGNED_MAIN_RING_INDEX, rankSize, txSliceIdxMain, rxSliceIdxMain, subSliceIdxMain, txReduceMemsMain,
163 : rxReduceMemsMain, localSrcMemsMain, localDstMemsMain));
164 : // 主环从流
165 0 : u32 sliceSize = multRingsSlices_[ALIGNED_MAIN_RING_INDEX].size() / rankSize;
166 0 : for (u32 memIdx = 0; memIdx < sliceSize; memIdx++) {
167 0 : CHK_RET(LocalMemcpy(
168 : step, rankSize, ALIGNED_MAIN_RING_INDEX, localSrcMemsMain[memIdx], localDstMemsMain[memIdx]));
169 : }
170 : // 更新索引
171 0 : subSliceIdxMain = (subSliceIdxMain + rankSize - 1) % rankSize;
172 0 : txSliceIdxMain = (txSliceIdxMain + rankSize - 1) % rankSize;
173 0 : rxSliceIdxMain = (rxSliceIdxMain + rankSize - 1) % rankSize;
174 0 : }
175 0 : return HCCL_SUCCESS;
176 : }
177 :
178 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::PreSync(const u32 ringIndex)
179 : {
180 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy] PreSync starts");
181 0 : if (ringIndex == 1) {
182 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB && (!disableDMAReduce_)) {
183 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[0], profilerInput_.stage));
184 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[1], profilerInput_.stage));
185 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
186 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[0], profilerInput_.stage));
187 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[1], profilerInput_.stage));
188 : } else {
189 0 : CHK_RET(MainWaitSub());
190 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
191 0 : CHK_RET(MainRecordSub());
192 : }
193 : } else {
194 0 : CHK_RET(LocalNotify::Post(subStreams_[0], dispatcher_, mainSignals_[0], profilerInput_.stage));
195 0 : CHK_RET(LocalNotify::Wait(subStreams_[0], dispatcher_, subSignals_[0], profilerInput_.stage));
196 : }
197 0 : return HCCL_SUCCESS;
198 : }
199 :
200 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::RunAllStreams(
201 : const u32 step, const u32 rankSize, std::vector<SenderMemoryInfo>& mainTxReduceMems,
202 : std::vector<ReducerMemoryInfo>& mainRxReduceMems, std::vector<SenderMemoryInfo>& subTxReduceMems,
203 : std::vector<ReducerMemoryInfo>& subRxReduceMems, std::vector<DeviceMem>& mainLocalSrcMems,
204 : std::vector<DeviceMem>& mainLocalDstMems, std::vector<DeviceMem>& subLocalSrcMems,
205 : std::vector<DeviceMem>& subLocalDstMems)
206 : {
207 : (void)mainLocalDstMems;
208 : (void)mainTxReduceMems;
209 : (void)subTxReduceMems;
210 : (void)mainLocalSrcMems;
211 0 : Stream mainStream;
212 0 : LINK mainPreLink;
213 0 : LINK mainNextLink;
214 0 : Stream subStream;
215 0 : LINK subPreLink;
216 0 : LINK subNextLink;
217 0 : CHK_RET(PrepareRunMainStream(ALIGNED_MAIN_RING_INDEX, mainStream, mainPreLink, mainNextLink));
218 0 : HCCL_DEBUG(
219 : "Reduce: step[%u] ring[%u], src rank[%u] starts to send slice to dst rank[%u]", step, ALIGNED_MAIN_RING_INDEX,
220 : mainPreLink->GetRemoteRank(), mainNextLink->GetRemoteRank());
221 0 : CHK_RET(PrepareRunMainStream(ALIGNED_SUB_RING_INDEX, subStream, subPreLink, subNextLink));
222 0 : HCCL_DEBUG(
223 : "Reduce: step[%u] ring[%u], src rank[%u] starts to send slice to dst rank[%u]", step, ALIGNED_SUB_RING_INDEX,
224 : subPreLink->GetRemoteRank(), subNextLink->GetRemoteRank());
225 :
226 0 : CHK_RET(mainNextLink->TxAck(mainStream));
227 0 : CHK_RET(mainPreLink->RxAck(mainStream));
228 0 : CHK_RET(subNextLink->TxAck(subStream));
229 0 : CHK_RET(subPreLink->RxAck(subStream));
230 :
231 0 : u32 sliceSize = multRingsSlices_[ALIGNED_MAIN_RING_INDEX].size() / rankSize;
232 0 : for (u32 memIdx = 0; memIdx < sliceSize; memIdx++) {
233 0 : CHK_RET(ReducerRun(ALIGNED_MAIN_RING_INDEX, dispatcher_, mainPreLink, mainRxReduceMems[memIdx], mainStream));
234 0 : CHK_RET(ReducerRun(ALIGNED_SUB_RING_INDEX, dispatcher_, subPreLink, subRxReduceMems[memIdx], subStream));
235 0 : CHK_RET(LocalMemcpy(step, rankSize, ALIGNED_SUB_RING_INDEX, subLocalSrcMems[memIdx], subLocalDstMems[memIdx]));
236 : }
237 0 : CHK_RET(mainPreLink->TxDataSignal(mainStream));
238 0 : CHK_RET(mainNextLink->RxDataSignal(mainStream));
239 0 : CHK_RET(subPreLink->TxDataSignal(subStream));
240 0 : CHK_RET(subNextLink->RxDataSignal(subStream));
241 0 : return HCCL_SUCCESS;
242 0 : }
243 :
244 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::RunReduceScatter(const u32 rank, const u32 rankSize)
245 : {
246 0 : HCCL_INFO("AlignedReduceScatterDoubleRingWithSerialLocalCopy starts, the input param rank[%u]", rank);
247 : // 空拷贝用于后续操作附着
248 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
249 : // 先完成主环主流操作
250 0 : CHK_RET(RunMainRingSubStream(rank, rankSize));
251 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
252 0 : CHK_RET(RunMainInitStep(rank, rankSize));
253 : // 主环主流通知从环主流开始通信
254 0 : CHK_RET(MainRecordSub());
255 : // 从环主流等待主环主流通知
256 0 : CHK_RET(SubWaitMain());
257 0 : CHK_RET(RunSubInitStep(rank, rankSize));
258 : // 从流通知主流通信完成
259 0 : CHK_RET(SubRecordMain());
260 : // 主流等待从流通知
261 0 : CHK_RET(MainWaitSub());
262 : }
263 : // 主环主流通知从环主流开始通信
264 0 : CHK_RET(MainRecordSub());
265 : // 从环主流等待主环主流通知
266 0 : CHK_RET(SubWaitMain());
267 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
268 0 : CHK_RET(ExecEmptyTasks());
269 : // 例如rank[0,1,2,3]中，rank0的rxSliceIdx = 2，txSliceIdx = 3, subSliceIdx = 1
270 : // 从环初始indexes
271 0 : u32 txSliceIdxSub = (rank + rankSize - 1) % rankSize;
272 0 : u32 rxSliceIdxSub = (rank + rankSize - DMA_REDUCE_TWO_OFFSET) % rankSize;
273 0 : u32 subSliceIdxSub = (rank + rankSize - DMA_REDUCE_THREE_OFFSET) % rankSize;
274 : // 主环初始indexes
275 0 : u32 txSliceIdxMain = (rankSize - rank - 1 + rankSize) % rankSize;
276 0 : u32 rxSliceIdxMain = (rankSize - rank - DMA_REDUCE_TWO_OFFSET + rankSize) % rankSize;
277 0 : u32 subSliceIdxMain = (rankSize - rank - DMA_REDUCE_THREE_OFFSET + rankSize) % rankSize;
278 :
279 0 : for (u32 step = 0; step < rankSize - 1; step++) {
280 : // 并发
281 0 : std::vector<SenderMemoryInfo> txReduceMemsMain;
282 0 : std::vector<ReducerMemoryInfo> rxReduceMemsMain;
283 0 : std::vector<SenderMemoryInfo> txReduceMemsSub;
284 0 : std::vector<ReducerMemoryInfo> rxReduceMemsSub;
285 0 : std::vector<DeviceMem> localSrcMemsMain;
286 0 : std::vector<DeviceMem> localDstMemsMain;
287 0 : std::vector<DeviceMem> localSrcMemsSub;
288 0 : std::vector<DeviceMem> localDstMemsSub;
289 0 : CHK_RET(PreRunStreams(
290 : step, rankSize, txSliceIdxMain, rxSliceIdxMain, subSliceIdxMain, txSliceIdxSub, rxSliceIdxSub,
291 : subSliceIdxSub, txReduceMemsMain, rxReduceMemsMain, txReduceMemsSub, rxReduceMemsSub, localSrcMemsMain,
292 : localDstMemsMain, localSrcMemsSub, localDstMemsSub));
293 0 : CHK_RET(RunAllStreams(
294 : step, rankSize, txReduceMemsMain, rxReduceMemsMain, txReduceMemsSub, rxReduceMemsSub, localSrcMemsMain,
295 : localDstMemsMain, localSrcMemsSub, localDstMemsSub));
296 : // 更新索引
297 0 : subSliceIdxSub = (subSliceIdxSub + rankSize - 1) % rankSize;
298 0 : txSliceIdxSub = (txSliceIdxSub + rankSize - 1) % rankSize;
299 0 : rxSliceIdxSub = (rxSliceIdxSub + rankSize - 1) % rankSize;
300 0 : subSliceIdxMain = (subSliceIdxMain + rankSize - 1) % rankSize;
301 0 : txSliceIdxMain = (txSliceIdxMain + rankSize - 1) % rankSize;
302 0 : rxSliceIdxMain = (rxSliceIdxMain + rankSize - 1) % rankSize;
303 0 : }
304 : // 从环主流通知主环主流通信完成
305 0 : CHK_RET(SubRecordMain());
306 : // 主环主流等待从环主流通知
307 0 : CHK_RET(MainWaitSub());
308 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
309 0 : HCCL_INFO("AlignedReduceScatterDoubleRingWithSerialLocalCopy finished to RunReduceScatter");
310 0 : return HCCL_SUCCESS;
311 : }
312 :
313 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::GetActiveSubstreamNumWithSerial(u32& activeSubstreamNum)
314 : {
315 0 : constexpr u32 IDX_2 = 2;
316 0 : activeSubstreamNum = subStreams_.size();
317 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB) {
318 0 : if (subStreams_.size() <= IDX_2) {
319 0 : HCCL_ERROR("[GetActiveSubstreamNumWithSerial]subStreams_.size()[%zu] <= 2", subStreams_.size());
320 0 : return HCCL_E_PARA;
321 : }
322 0 : if (disableDMAReduce_) {
323 0 : activeSubstreamNum = subStreams_.size() - IDX_2;
324 : } else {
325 0 : activeSubstreamNum = subStreams_.size() - 1;
326 : }
327 0 : HCCL_DEBUG(
328 : "[AlignedReduceScatterDoubleRingWithSerialLocalCopy] disableDMAReduce_[%d], activeSubstreamNum[%u]",
329 : disableDMAReduce_, activeSubstreamNum);
330 : }
331 0 : return HCCL_SUCCESS;
332 : }
333 :
334 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::ExecEmptyTasks()
335 : {
336 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy] ExecEmptyTasks");
337 0 : u32 activeSubstreamNum = 0;
338 0 : CHK_RET(GetActiveSubstreamNumWithSerial(activeSubstreamNum));
339 0 : for (u32 signalIndex = 0; signalIndex < activeSubstreamNum; signalIndex++) {
340 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, subStreams_[signalIndex], dispatcher_));
341 : }
342 0 : return HCCL_SUCCESS;
343 : }
344 :
345 : // 主流通知从流干活
346 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::MainRecordSub()
347 : {
348 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy] MainRecordSub");
349 0 : u32 activeSubstreamNum = 0;
350 0 : CHK_RET(GetActiveSubstreamNumWithSerial(activeSubstreamNum));
351 0 : for (u32 signalIndex = 0; signalIndex < activeSubstreamNum; signalIndex++) {
352 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[signalIndex], profilerInput_.stage));
353 : }
354 0 : return HCCL_SUCCESS;
355 : }
356 : // 从流等待主流
357 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::SubWaitMain()
358 : {
359 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy] SubWaitMain");
360 0 : u32 activeSubstreamNum = 0;
361 0 : CHK_RET(GetActiveSubstreamNumWithSerial(activeSubstreamNum));
362 0 : for (u32 streamIndex = 0; streamIndex < activeSubstreamNum; streamIndex++) {
363 0 : CHK_RET(
364 : LocalNotify::Wait(subStreams_[streamIndex], dispatcher_, subSignals_[streamIndex], profilerInput_.stage));
365 : }
366 0 : return HCCL_SUCCESS;
367 : }
368 : // 主流等待从流
369 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::MainWaitSub()
370 : {
371 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy] MainWaitSub");
372 0 : u32 activeSubstreamNum = 0;
373 0 : CHK_RET(GetActiveSubstreamNumWithSerial(activeSubstreamNum));
374 0 : for (u32 signalIndex = 0; signalIndex < activeSubstreamNum; signalIndex++) {
375 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[signalIndex], profilerInput_.stage));
376 : }
377 0 : return HCCL_SUCCESS;
378 : }
379 : // 从流告诉主流活干完了
380 0 : HcclResult AlignedReduceScatterDoubleRingWithSerialLocalCopy::SubRecordMain()
381 : {
382 0 : HCCL_DEBUG("[AlignedReduceScatterDoubleRingWithSerialLocalCopy] SubRecordMain");
383 0 : u32 activeSubstreamNum = 0;
384 0 : CHK_RET(GetActiveSubstreamNumWithSerial(activeSubstreamNum));
385 0 : for (u32 streamIndex = 0; streamIndex < activeSubstreamNum; streamIndex++) {
386 0 : CHK_RET(
387 : LocalNotify::Post(subStreams_[streamIndex], dispatcher_, mainSignals_[streamIndex], profilerInput_.stage));
388 : }
389 0 : return HCCL_SUCCESS;
390 : }
391 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_REDUCESCATTER_DB_RING_SLC, AlignedReduceScatterDoubleRingWithSerialLocalCopy);
392 : } // namespace hccl
|
