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