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 "all_gather_ring_concurrent_direct.h"
12 : #include "alg_template_register.h"
13 :
14 : namespace hccl {
15 0 : AllGatherRingConcurrentDirect::AllGatherRingConcurrentDirect(const HcclDispatcher dispatcher)
16 0 : : AlgTemplateBase(dispatcher)
17 0 : {}
18 :
19 0 : AllGatherRingConcurrentDirect::~AllGatherRingConcurrentDirect() {}
20 :
21 0 : HcclResult AllGatherRingConcurrentDirect::Prepare(
22 : HcomCollOpInfo* opInfo, const u32 userRank, std::vector<Stream>& subStreams,
23 : const std::vector<std::shared_ptr<LocalNotify>>& mainSignals,
24 : const std::vector<std::shared_ptr<LocalNotify>>& subSignals, const std::vector<u32>& ringsOrder,
25 : const std::vector<Slice>& userMemSlices, bool isSdma)
26 : {
27 0 : opInfo_ = opInfo;
28 0 : userRank_ = userRank;
29 0 : subStreams_ = subStreams;
30 0 : mainSignals_ = mainSignals;
31 0 : subSignals_ = subSignals;
32 0 : ringsOrder_ = ringsOrder;
33 0 : userMemOutputSlices_ = userMemSlices;
34 0 : isSdma_ = isSdma;
35 0 : return HCCL_SUCCESS;
36 : }
37 :
38 : // 服务器间allgather的入口函数
39 0 : HcclResult AllGatherRingConcurrentDirect::RunAsync(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
40 : {
41 : // 基本的检查
42 0 : CHK_RET(CheckParameters(rank, rankSize, links));
43 :
44 0 : if (rankSize == 1) {
45 0 : CHK_RET(OneRankMemcpy());
46 0 : return HCCL_SUCCESS;
47 : }
48 : // 收集邻居信息
49 0 : CHK_RET(GetInitializedNeighborLinks(rank, rankSize, links));
50 :
51 : // 填充slice_
52 0 : CHK_RET(SetSlices(rank, rankSize));
53 :
54 : // 运行all-gather, ring算法
55 0 : CHK_RET(RunAllGather(rank, rankSize));
56 :
57 0 : if (barrierSwitchOn_) {
58 : // 执行barrier，保证数据收发完成
59 0 : CHK_RET(ExecuteBarrier(leftLink_, rightLink_));
60 : }
61 :
62 0 : CHK_RET(LaunchTaskExtend(dispatcher_, stream_, subStreams_));
63 :
64 0 : HCCL_INFO("AllGatherRingConcurrentDirect finished: rank[%u] end", rank);
65 0 : return HCCL_SUCCESS;
66 : }
67 :
68 : HcclResult
69 0 : AllGatherRingConcurrentDirect::CheckParameters(const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
70 : {
71 0 : CHK_PTR_NULL(opInfo_);
72 0 : CHK_RET(CheckConcurrentDirectParameters(rank, rankSize, links));
73 : // 判断subStreams数量是否正确
74 0 : CHK_PRT_RET(
75 : subStreams_.size() < 1,
76 : HCCL_ERROR("[AllGatherRingConcurrentDirect] subStreams size[%u] is less than 1", subStreams_.size()),
77 : HCCL_E_PARA);
78 0 : for (auto& s : subStreams_) {
79 0 : CHK_PTR_NULL(s.ptr());
80 : }
81 : // 判断mainSignals数量是否正确
82 0 : CHK_PRT_RET(
83 : mainSignals_.size() < 1,
84 : HCCL_ERROR("[AllGatherRingConcurrentDirect] mainSignals size[%u] is less than 1", mainSignals_.size()),
85 : HCCL_E_PARA);
86 : // 判断subSignals数量是否正确
87 0 : CHK_PRT_RET(
88 : subSignals_.size() < 1,
89 : HCCL_ERROR("[AllGatherRingConcurrentDirect] subSignals size[%u] is less than 1", subSignals_.size()),
90 : HCCL_E_PARA);
91 : // 判断ringsOrder数量是否正确
92 0 : CHK_PRT_RET(
93 : ringsOrder_.size() % rankSize != 0,
94 : HCCL_ERROR(
95 : "[AllGatherRingConcurrentDirect] ringsOrder size[%u] can not be divided by rank size[%u]",
96 : ringsOrder_.size(), rankSize),
97 : HCCL_E_PARA);
98 : // 判断userMemInputSlices数量是否正确
99 0 : CHK_PRT_RET(
100 : userMemOutputSlices_.size() % rankSize != 0,
101 : HCCL_ERROR(
102 : "[AllGatherRingConcurrentDirect] userMemOutputSlices size[%u] can not be divided by rank size[%u]",
103 : userMemOutputSlices_.size(), rankSize),
104 : HCCL_E_PARA);
105 0 : HCCL_INFO("AllGatherRingConcurrentDirect finished to CheckParameters");
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 0 : HcclResult AllGatherRingConcurrentDirect::OneRankMemcpy()
110 : {
111 0 : for (u32 sliceIdx = 0; sliceIdx < slices_.size(); sliceIdx++) {
112 0 : const Slice& srcSlice = slices_[sliceIdx];
113 0 : const Slice& dstSlice = userMemOutputSlices_[sliceIdx];
114 0 : DeviceMem src;
115 0 : DeviceMem dst = DeviceMem::create(static_cast<u8*>(opInfo_->outputAddr) + dstSlice.offset, dstSlice.size);
116 0 : if (opInfo_->inputAddr != nullptr) {
117 : // opInfo_->inputAddr != nullptr指示要从user input获取输入
118 0 : u64 stepOffset = slices_[ringsOrder_[0]].offset;
119 0 : HCCL_DEBUG(
120 : "Memcpy operation: stream[main], rank[%u] starts to copy offset[%llu], size[%llu] at userInput",
121 : userRank_, stepOffset, srcSlice.size);
122 0 : src = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + stepOffset, srcSlice.size);
123 : } else {
124 : // opInfo_->inputAddr == nullptr指示要从CCL buffer获取输入
125 0 : HCCL_DEBUG(
126 : "Memcpy operation: stream[main], rank[%u] starts to copy offset[%llu], size[%llu] at inputMem_",
127 : userRank_, srcSlice.offset, srcSlice.size);
128 0 : src = inputMem_.range(srcSlice.offset, srcSlice.size);
129 : }
130 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, stream_));
131 0 : HCCL_DEBUG("[AllGatherRingConcurrentDirect][OneRankMemcpy]sliceIdx[%u] for Memcpy success", sliceIdx);
132 0 : }
133 :
134 0 : return HCCL_SUCCESS;
135 : }
136 :
137 0 : HcclResult AllGatherRingConcurrentDirect::GetInitializedNeighborLinks(
138 : const u32 rank, const u32 rankSize, const std::vector<LINK>& links)
139 : {
140 : // 收集左邻居信息
141 0 : leftLink_ = links[(rank + rankSize - 1) % rankSize];
142 0 : CHK_SMART_PTR_NULL(leftLink_);
143 :
144 : // 收集右邻居信息
145 0 : rightLink_ = links[(rank + 1) % rankSize];
146 0 : CHK_SMART_PTR_NULL(rightLink_);
147 0 : HCCL_INFO("AllGatherRingConcurrentDirect finished to GetInitializedNeighborLinks");
148 0 : return HCCL_SUCCESS;
149 : }
150 :
151 0 : HcclResult AllGatherRingConcurrentDirect::SetSlices(const u32 rank, const u32 rankSize)
152 : {
153 0 : inputSlices_ = slices_;
154 0 : if (slices_.size() == 0) {
155 0 : slices_.resize(rankSize);
156 0 : inputSlices_.resize(rankSize);
157 :
158 0 : u64 sliceSize = count_ * DataUnitSize(dataType_);
159 0 : for (u32 i = 0; i < rankSize; i++) {
160 0 : slices_[i].size = sliceSize;
161 0 : slices_[i].offset = sliceSize * i;
162 0 : inputSlices_[i].size = sliceSize;
163 0 : inputSlices_[i].offset = (inputMem_.size() < outputMem_.size()) ? 0 : (sliceSize * i);
164 0 : HCCL_DEBUG(
165 : "rank[%u], slices[%u].offset=%llu, slices[%u].size=[%llu]", rank, i, slices_[i].offset, i,
166 : slices_[i].size);
167 : }
168 : }
169 0 : if (UNLIKELY(HcclCheckLogLevel(DLOG_DEBUG))) {
170 0 : for (u32 i = 0; i < slices_.size(); i++) {
171 0 : HCCL_DEBUG(
172 : "[AllGatherRingConcurrentDirect][SetSlices]rank[%u], slices[%u].offset=[%llu], slices[%u].size=[%llu]",
173 : rank, i, slices_[i].offset, i, slices_[i].size);
174 : }
175 : }
176 0 : HCCL_INFO("AllGatherRingConcurrentDirect finished to SetSlices");
177 0 : return HCCL_SUCCESS;
178 : }
179 :
180 0 : HcclResult AllGatherRingConcurrentDirect::RunInitStep(const u32 rank, const u32 rankSize)
181 : {
182 : // 第一步搬到userMemIn_的offset, 不同的ring环offset不一样
183 0 : auto firstStepOffset = slices_[ringsOrder_[0]].offset;
184 : // 第-1步，片内将部分数据从userIn搬到cclIn
185 0 : DeviceMem srcInit;
186 0 : DeviceMem dstInit;
187 0 : u32 initSliceIdx = rank;
188 0 : u32 sliceSize = slices_.size() / rankSize;
189 0 : for (u32 sliceIdx = 0; sliceIdx < sliceSize; sliceIdx++) {
190 0 : Slice initSlice = slices_[initSliceIdx * sliceSize + sliceIdx];
191 : // 需要+userMemIn_的offset
192 0 : if (opInfo_->inputAddr != nullptr) {
193 : // AllGather算子调用AllGatherRingConcurrentDirect场景
194 0 : srcInit = DeviceMem::create(static_cast<u8*>(opInfo_->inputAddr) + firstStepOffset, initSlice.size);
195 : } else {
196 : // AllReduce算子调用AllGatherRingConcurrentDirect场景
197 0 : srcInit = inputMem_.range(initSlice.offset, initSlice.size);
198 : }
199 0 : dstInit = outputMem_.range(initSlice.offset, initSlice.size);
200 0 : HCCL_DEBUG(
201 : "Memcpy operation: step[-1] stream[main] src rank[%u] starts to copy(rcv) offset[%llu], "
202 : "size[%llu] on userMemOutput to offset[%llu], size[%llu] on CCL",
203 : userRank_, firstStepOffset, initSlice.size, initSlice.offset, initSlice.size);
204 : // 若src与dst一样，则不需要搬运
205 0 : if (srcInit != dstInit) {
206 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstInit, srcInit, stream_));
207 : }
208 : }
209 0 : return HCCL_SUCCESS;
210 0 : }
211 :
212 : // 从流单个slice的拷贝任务：本端cclout -> 本端userout
213 0 : HcclResult AllGatherRingConcurrentDirect::RunSubStreamSlice(
214 : const u32 step, const u32 sliceIdx, const std::vector<Slice>& txSliceVector,
215 : const std::vector<Slice>& subSliceVector)
216 : {
217 0 : DeviceMem src = outputMem_.range(txSliceVector[sliceIdx].offset, txSliceVector[sliceIdx].size);
218 : DeviceMem dst = DeviceMem::create(
219 0 : static_cast<u8*>(opInfo_->outputAddr) + subSliceVector[sliceIdx].offset, subSliceVector[sliceIdx].size);
220 0 : HCCL_DEBUG(
221 : "Memcpy operation: step[%u] stream[sub], src rank[%u] starts to send offset[%llu] size[%llu], "
222 : "dst rank starts to rcv offset[%llu] size[%llu] at userMemOutput_",
223 : step, userRank_, subSliceVector[sliceIdx].offset, subSliceVector[sliceIdx].size, txSliceVector[sliceIdx].offset,
224 : txSliceVector[sliceIdx].size);
225 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dst, src, subStreams_[0]));
226 0 : return HCCL_SUCCESS;
227 0 : }
228 :
229 0 : HcclResult AllGatherRingConcurrentDirect::RunAllGather(const u32 rank, const u32 rankSize)
230 : {
231 0 : HCCL_INFO("AllGatherRingConcurrentDirect starts, the input param rank[%u]", rank);
232 : // 空拷贝用于后续操作附着
233 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
234 :
235 0 : CHK_RET(RunInitStep(rank, rankSize));
236 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
237 0 : CHK_RET(MainRecordSub()); // 主流通知从流开始通信
238 0 : CHK_RET(SubWaitMain()); // 从流等待主流通知
239 : // 空拷贝用于主从流任务并发
240 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
241 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, subStreams_[0], dispatcher_));
242 0 : u32 txSliceIdx = rank;
243 0 : u32 sliceSize = slices_.size() / rankSize;
244 0 : u32 rxSliceIdx = (rank + rankSize - 1) % rankSize;
245 :
246 0 : std::vector<DeviceMem> finalSrc;
247 0 : std::vector<DeviceMem> finalDst;
248 0 : for (u32 step = 0; step < rankSize - 1; step++) {
249 0 : std::vector<Slice> rxSliceVector;
250 0 : std::vector<Slice> mainSliceVector;
251 0 : std::vector<Slice> txSliceVector;
252 0 : std::vector<Slice> subSliceVector;
253 0 : for (u32 sliceIdx = 0; sliceIdx < sliceSize; sliceIdx++) {
254 0 : rxSliceVector.push_back(slices_[rxSliceIdx * sliceSize + sliceIdx]);
255 0 : mainSliceVector.push_back(userMemOutputSlices_[rxSliceIdx * sliceSize + sliceIdx]);
256 0 : txSliceVector.push_back(slices_[txSliceIdx * sliceSize + sliceIdx]);
257 0 : subSliceVector.push_back(userMemOutputSlices_[txSliceIdx * sliceSize + sliceIdx]);
258 : }
259 : // 主流
260 : // Ack
261 0 : CHK_RET(leftLink_->TxAck(stream_));
262 0 : CHK_RET(rightLink_->RxAck(stream_));
263 :
264 0 : std::vector<TxMemoryInfo> txMems;
265 0 : std::vector<RxMemoryInfo> rxMems;
266 0 : for (u32 sliceIdx = 0; sliceIdx < sliceSize; sliceIdx++) {
267 0 : DeviceMem src = outputMem_.range(txSliceVector[sliceIdx].offset, txSliceVector[sliceIdx].size);
268 0 : HCCL_DEBUG(
269 : "tx srcMem[%p] range[%llu] size[%llu] ", src.ptr(), txSliceVector[sliceIdx].offset,
270 : txSliceVector[sliceIdx].size);
271 0 : txMems.emplace_back(TxMemoryInfo{
272 0 : UserMemType::OUTPUT_MEM, txSliceVector[sliceIdx].offset + baseOffset_, src.ptr(),
273 0 : txSliceVector[sliceIdx].size});
274 0 : DeviceMem dst;
275 0 : if (isSdma_ && step == rankSize - DMA_REDUCE_TWO_OFFSET) {
276 0 : HCCL_DEBUG(
277 : "DMAReduce(sdma) MemcpyAsync operation: step[%u] stream[main], dst rank[%u] starts to rcv "
278 : "offset[%llu] size[%llu] at userMemOutput_",
279 : step, userRank_, mainSliceVector[sliceIdx].offset, mainSliceVector[sliceIdx].size);
280 0 : dst = DeviceMem::create(
281 0 : static_cast<u8*>(opInfo_->outputAddr) + mainSliceVector[sliceIdx].offset,
282 0 : mainSliceVector[sliceIdx].size);
283 : } else {
284 0 : HCCL_DEBUG(
285 : "MemcpyAsync operation: step[%u] stream[main], dst rank[%u] starts to rcv offset[%llu] size[%llu] "
286 : "at outputMem_",
287 : step, userRank_, rxSliceVector[sliceIdx].offset, rxSliceVector[sliceIdx].size);
288 0 : dst = outputMem_.range(rxSliceVector[sliceIdx].offset, rxSliceVector[sliceIdx].size);
289 0 : if (!isSdma_ && step == rankSize - DMA_REDUCE_TWO_OFFSET) {
290 0 : HCCL_DEBUG("DMAReduce(rdma) record final addr");
291 0 : finalSrc.push_back(outputMem_.range(rxSliceVector[sliceIdx].offset, rxSliceVector[sliceIdx].size));
292 0 : finalDst.push_back(DeviceMem::create(
293 0 : static_cast<u8*>(opInfo_->outputAddr) + mainSliceVector[sliceIdx].offset,
294 0 : mainSliceVector[sliceIdx].size));
295 : }
296 : }
297 0 : rxMems.emplace_back(RxMemoryInfo{
298 0 : UserMemType::OUTPUT_MEM, rxSliceVector[sliceIdx].offset + baseOffset_, dst.ptr(),
299 0 : rxSliceVector[sliceIdx].size});
300 0 : }
301 0 : CHK_RET(rightLink_->TxAsync(txMems, stream_));
302 :
303 : // dispatcher_aicpu 单条流的任务队列存在上限，队列满后host会阻塞下发，因此主流与从流的任务必须
304 : // 交替下发：从流Wait(subSignals)依赖主流Post(subSignals)，主流Wait(mainSignals)依赖从流
305 : // Post(mainSignals)。若先集中下发某一条流的全部任务，队列被占满后host阻塞，而队列中等待的信号
306 : // 又需要另一条流尚未下发的任务来产生，两条流互相死等。以下保证每个Wait与其配对的Post在小窗口
307 : // 内先后完成下发，且每条流上的任务序列保持不变。
308 0 : if (!isSdma_) {
309 : // 从流先Post(mainSignals)，主流Wait/Empty/Post与RxAsync下发完成后，从流再Wait并下发本步拷贝任务
310 0 : CHK_RET(LocalNotify::Post(subStreams_[0], dispatcher_, mainSignals_[0], profilerInput_.stage));
311 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[0], profilerInput_.stage));
312 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
313 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[0], profilerInput_.stage));
314 0 : CHK_RET(leftLink_->RxAsync(rxMems, stream_));
315 0 : CHK_RET(LocalNotify::Wait(subStreams_[0], dispatcher_, subSignals_[0], profilerInput_.stage));
316 : // 从流：本端cclout -> 本端userout
317 0 : for (u32 sliceIdx = 0; sliceIdx < sliceSize; sliceIdx++) {
318 0 : CHK_RET(RunSubStreamSlice(step, sliceIdx, txSliceVector, subSliceVector));
319 : }
320 : } else {
321 0 : CHK_RET(leftLink_->RxDataSignal(stream_));
322 : // 每个slice按 从流Post -> 主流Wait/Empty/Post -> 从流Wait -> 从流拷贝 -> 主流远端读 的顺序交替下发
323 0 : for (u32 sliceIdx = 0; sliceIdx < sliceSize; sliceIdx++) {
324 0 : CHK_RET(LocalNotify::Post(subStreams_[0], dispatcher_, mainSignals_[0], profilerInput_.stage));
325 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[0], profilerInput_.stage));
326 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem_, outputMem_, stream_, dispatcher_));
327 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[0], profilerInput_.stage));
328 0 : CHK_RET(LocalNotify::Wait(subStreams_[0], dispatcher_, subSignals_[0], profilerInput_.stage));
329 : // 从流：本端cclout -> 本端userout
330 0 : CHK_RET(RunSubStreamSlice(step, sliceIdx, txSliceVector, subSliceVector));
331 : // 主流：对端cclout -> 本端（远端读）
332 0 : auto& mem = rxMems[sliceIdx];
333 0 : CHK_PTR_NULL(mem.dst);
334 0 : void* srcMemPtr = nullptr;
335 0 : CHK_RET(leftLink_->GetRemoteMem(mem.srcMemType, &srcMemPtr));
336 :
337 0 : DeviceMem srcDevMem(static_cast<s8*>(srcMemPtr) + mem.srcOffset, mem.len);
338 0 : DeviceMem dstDevMem(static_cast<s8*>(mem.dst), mem.len);
339 0 : CHK_RET(HcclD2DMemcpyAsync(
340 : dispatcher_, dstDevMem, srcDevMem, stream_, leftLink_->GetRemoteRank(), leftLink_->GetLinkType()));
341 0 : }
342 : }
343 :
344 : // 更新索引
345 0 : txSliceIdx = (txSliceIdx + rankSize - 1) % rankSize;
346 0 : rxSliceIdx = (rxSliceIdx + rankSize - 1) % rankSize;
347 0 : }
348 0 : CHK_RET(SubRecordMain()); // 从流通知主流通信完成
349 0 : CHK_RET(MainWaitSub()); // 主流等待从流通知
350 0 : if (!isSdma_) {
351 0 : for (u32 vecIdx = 0; vecIdx < finalSrc.size(); vecIdx++) {
352 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, finalDst[vecIdx], finalSrc[vecIdx], stream_));
353 : }
354 : }
355 0 : HCCL_INFO("AllGatherRingConcurrentDirect finished to RunAllGather");
356 0 : return HCCL_SUCCESS;
357 0 : }
358 :
359 : // 主流通知从流干活
360 0 : HcclResult AllGatherRingConcurrentDirect::MainRecordSub()
361 : {
362 0 : for (u32 signalIndex = 0; signalIndex < subSignals_.size(); signalIndex++) {
363 0 : CHK_RET(LocalNotify::Post(stream_, dispatcher_, subSignals_[signalIndex], profilerInput_.stage));
364 : }
365 0 : return HCCL_SUCCESS;
366 : }
367 : // 从流等待主流
368 0 : HcclResult AllGatherRingConcurrentDirect::SubWaitMain()
369 : {
370 0 : for (u32 streamIndex = 0; streamIndex < subSignals_.size(); streamIndex++) {
371 0 : CHK_RET(
372 : LocalNotify::Wait(subStreams_[streamIndex], dispatcher_, subSignals_[streamIndex], profilerInput_.stage));
373 : }
374 0 : return HCCL_SUCCESS;
375 : }
376 : // 主流等待从流
377 0 : HcclResult AllGatherRingConcurrentDirect::MainWaitSub()
378 : {
379 0 : for (u32 signalIndex = 0; signalIndex < mainSignals_.size(); signalIndex++) {
380 0 : CHK_RET(LocalNotify::Wait(stream_, dispatcher_, mainSignals_[signalIndex], profilerInput_.stage));
381 : }
382 0 : return HCCL_SUCCESS;
383 : }
384 : // 从流告诉主流活干完了
385 0 : HcclResult AllGatherRingConcurrentDirect::SubRecordMain()
386 : {
387 0 : for (u32 streamIndex = 0; streamIndex < mainSignals_.size(); streamIndex++) {
388 0 : CHK_RET(
389 : LocalNotify::Post(subStreams_[streamIndex], dispatcher_, mainSignals_[streamIndex], profilerInput_.stage));
390 : }
391 0 : return HCCL_SUCCESS;
392 : }
393 : REGISTER_TEMPLATE(TemplateType::TEMPLATE_ALL_GATHER_RING_CONCURRENT_DIRECT, AllGatherRingConcurrentDirect);
394 : } // namespace hccl
|
