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 "coll_all_gather_ring_zerocopy_pipeline_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllGatherRingZerocopyPipelineExecutor::CollAllGatherRingZerocopyPipelineExecutor(const HcclDispatcher dispatcher,
15 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
16 0 : : CollAllGatherExecutor(dispatcher, topoMatcher)
17 : {
18 0 : DMAReduceFlag_ = true; // 设为true，以禁用RunLoop中的本地拷贝
19 0 : desc_.isZeroCopy = true;
20 0 : desc_.level1SupportedAlgos = {
21 : AlgTypeLevel1::ALG_LEVEL1_NHR,
22 : AlgTypeLevel1::ALG_LEVEL1_NB,
23 : AlgTypeLevel1::ALG_LEVEL1_RING
24 0 : };
25 0 : desc_.level2SupportedAlgos = {
26 : AlgTypeLevel2::ALG_LEVEL2_PIPELINE
27 0 : };
28 0 : }
29 :
30 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcStreamNum(u32& streamNum)
31 : {
32 0 : u32 totalStreamNum = 1 + (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ?
33 : (LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE + 1) : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
34 0 : streamNum = totalStreamNum - 1;
35 0 : HCCL_INFO("[%s] tag[%s] streamNum[%u]", __func__, tag_.c_str(), streamNum);
36 0 : return HCCL_SUCCESS;
37 : }
38 :
39 0 : u64 CollAllGatherRingZerocopyPipelineExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
40 : {
41 0 : u32 serverNumPerSuperPod = topoAttr_.serverNum / topoAttr_.superPodNum;
42 0 : u32 bufferSliceNum = std::max(2U, serverNumPerSuperPod);
43 0 : u64 maxCountPerLoop = cclBuffSize / bufferSliceNum / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
44 0 : HCCL_INFO("[%s] tag[%s] maxCountPerLoop[%u]", __func__, tag_.c_str(), maxCountPerLoop);
45 0 : return maxCountPerLoop;
46 : }
47 :
48 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
49 : {
50 0 : TransportMemType inputType = TransportMemType::RESERVED;
51 0 : TransportMemType outputType = TransportMemType::RESERVED;
52 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
53 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
54 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
55 0 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
56 : // 额外增加数据交换的建链
57 0 : CHK_RET(CalcExchangeCommInfo(opTransport));
58 0 : return HCCL_SUCCESS;
59 : }
60 :
61 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcTransportMemType(TransportMemType &inputType,
62 : TransportMemType &outputType)
63 : {
64 0 : inputType = TransportMemType::CCL_INPUT;
65 0 : outputType = TransportMemType::CCL_OUTPUT;
66 0 : return HCCL_SUCCESS;
67 : }
68 :
69 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcLevel0CommInfo(TransportMemType inputType,
70 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
71 : {
72 0 : CommParaInfo commParaLevel0(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
73 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaLevel0, opTransport[COMM_LEVEL0], inputType, outputType));
74 0 : LevelNSubCommTransport &commTransportLevel0 = opTransport[COMM_LEVEL0];
75 0 : for (u32 subCommIndex = 0; subCommIndex < commTransportLevel0.size(); subCommIndex++) {
76 0 : commTransportLevel0[subCommIndex].isZeroCopy = true;
77 : }
78 0 : return HCCL_SUCCESS;
79 0 : }
80 :
81 : // PipeLine模式下使用Ring算法
82 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcLevel2CommInfo(TransportMemType inputType,
83 : TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
84 : {
85 0 : CommParaInfo commParaInfo(COMM_LEVEL2, CommType::COMM_TAG_RING_INNER);
86 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL2], inputType, outputType));
87 0 : return HCCL_SUCCESS;
88 0 : }
89 :
90 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcExchangeCommInfo(
91 : std::vector<LevelNSubCommTransport>& opTransport)
92 : {
93 0 : std::set<u32> commTargetUserRankSet;
94 0 : u32 remoteRankSend = 0;
95 0 : u32 remoteRankRecv = 0;
96 0 : CHK_RET(CalExchangeRemoteRank(remoteRankSend, remoteRankRecv));
97 0 : commTargetUserRankSet.insert(remoteRankSend);
98 0 : commTargetUserRankSet.insert(remoteRankRecv);
99 : CommParaInfo commParaInfo(COMM_COMBINE_ORDER, CommType::COMM_TAG_PARTIAL_MESH_COMBINED, INVALID_VALUE_RANKID,
100 0 : INVALID_VALUE_RANKID, false, false, commTargetUserRankSet);
101 :
102 0 : TransportMemType inputType = TransportMemType::CCL_INPUT;
103 0 : TransportMemType outputType = TransportMemType::CCL_OUTPUT;
104 :
105 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_COMBINE_ORDER], inputType, outputType));
106 0 : LevelNSubCommTransport &commTransport = opTransport[COMM_COMBINE_ORDER];
107 0 : for (u32 subCommIndex = 0; subCommIndex < commTransport.size(); subCommIndex++) {
108 0 : for (auto &transportRequest : commTransport[subCommIndex].transportRequests) {
109 0 : transportRequest.isUsedRdma = topoAttr_.isUsedRdmaMap.at(transportRequest.remoteUserRank);
110 : }
111 : }
112 0 : return HCCL_SUCCESS;
113 0 : }
114 :
115 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
116 : {
117 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherRingZerocopyPipelineExecutor][Orchestrate] begins.");
118 :
119 0 : HcclUs startut = TIME_NOW();
120 0 : tag_ = param.tag;
121 0 : algResResp_ = &algRes;
122 :
123 0 : mainStream_ = param.stream;
124 0 : subStreams_ = algResResp_->slaveStreams;
125 0 : rdmaMainStream_ = subStreams_.back();
126 0 : sdmaSubStreams_.assign(subStreams_.begin(), subStreams_.end() - 1);
127 0 : notifyMainToRdma_ = algResResp_->notifiesAux.back(); // 主流通知从流使用Aux
128 0 : notifyRdmaToMain_ = algResResp_->notifiesMain.back(); // 从流通知主流使用Main
129 0 : notifySdmaMain_.assign(algResResp_->notifiesMain.begin(), algResResp_->notifiesMain.end() - 1);
130 0 : notifySdmaSub_.assign(algResResp_->notifiesAux.begin(), algResResp_->notifiesAux.end() - 1);
131 :
132 0 : CHK_RET(GetCommRankInfoNormal(level0Rank_, level0RankSize_, level1Rank_, level1RankSize_, level2Rank_,
133 : level2RankSize_, false));
134 0 : unitSize_ = SIZE_TABLE[param.DataDes.dataType];
135 0 : totalSize_ = param.DataDes.count * unitSize_;
136 0 : blockSize_ = totalSize_ * level0RankSize_ * level1RankSize_;
137 :
138 0 : CHK_RET(RunLoop(param));
139 :
140 0 : HCCL_INFO("tag[%s], Allgather executor orchestrate success, take time [%lld]us.", tag_.c_str(),
141 : DURATION_US(TIME_NOW() - startut));
142 0 : return HCCL_SUCCESS;
143 : }
144 :
145 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::RunLoop(OpParam ¶m)
146 : {
147 0 : u8* curInputPtr = static_cast<u8 *>(param.inputPtr);
148 0 : u8* curOutputPtr = static_cast<u8 *>(param.outputPtr);
149 0 : CHK_PTR_NULL(curInputPtr);
150 0 : CHK_PTR_NULL(curOutputPtr);
151 :
152 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algResResp_->cclInputMem.size(), unitSize_);
153 0 : CHK_PRT_RET(maxCountPerLoop == 0,
154 : HCCL_ERROR("[CollAllGatherRingZerocopyPipelineExecutor][RunLoop]tag[%s] userRankSize[%u] maxCountPerLoop[%llu]",
155 : tag_.c_str(), topoAttr_.userRankSize, maxCountPerLoop), HCCL_E_PARA);
156 :
157 0 : u32 bufferLoopNum = (param.DataDes.count + maxCountPerLoop - 1) / maxCountPerLoop;
158 0 : u64 countLeft = param.DataDes.count;
159 0 : for (u32 loopIdx = 0; loopIdx < bufferLoopNum; loopIdx++) {
160 0 : bool isLastLoop = (loopIdx == bufferLoopNum - 1);
161 0 : u64 curCount = countLeft > maxCountPerLoop ? maxCountPerLoop : countLeft;
162 0 : countLeft -= curCount;
163 0 : u64 curSize = curCount * unitSize_;
164 :
165 0 : ExecMem execMem;
166 0 : execMem.count = curCount;
167 0 : execMem.inputMem = algResResp_->cclInputMem;
168 0 : execMem.outputMem = algResResp_->cclOutputMem;
169 0 : execMem.inputPtr = curInputPtr;
170 0 : execMem.outputPtr = curOutputPtr;
171 0 : CHK_RET(KernelRunWithLoop(param, execMem, isLastLoop));
172 0 : CHK_RET(LaunchTaskExtend(dispatcher_, mainStream_, subStreams_));
173 :
174 0 : curInputPtr += curSize;
175 0 : curOutputPtr += curSize;
176 0 : }
177 :
178 0 : return HCCL_SUCCESS;
179 : }
180 :
181 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::KernelRunWithLoop(const OpParam ¶m, ExecMem &execMem, bool isLastLoop)
182 : {
183 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherRingZerocopyPipelineExecutor][KernelRunWithLoop]KernelRun begins.");
184 0 : u64 curSize = execMem.count * unitSize_;
185 :
186 : // Local Copy: UserIn -> Ccl
187 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(execMem.inputPtr), curSize);
188 0 : DeviceMem dstMem = execMem.inputMem.range(0, curSize);
189 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, mainStream_));
190 :
191 0 : memIdx_ = 0;
192 0 : blockIdx_ = level2Rank_;
193 :
194 0 : for (u32 step = 0; step < level2RankSize_; step++) { // level2 ring算法的步数 + 结尾一步
195 : // 启动L2 RDMA
196 0 : if (step < level2RankSize_ - 1) {
197 0 : CHK_RET(NotifyRdmaStreamStart());
198 : }
199 :
200 : // L1 SDMA
201 0 : CHK_RET(KernelRunInterServer(param, execMem));
202 :
203 : // L0 Zcopy 最后一次ccl循环时，需要启动Server内零拷贝步骤
204 0 : if (isLastLoop) {
205 0 : u64 inputSize = totalSize_ * level1RankSize_;
206 0 : u8* inputPtr = static_cast<u8 *>(param.outputPtr) + blockSize_ * blockIdx_ + level0Rank_ * inputSize;
207 0 : u8* outputPtr = static_cast<u8 *>(param.outputPtr) + blockSize_ * blockIdx_;
208 0 : ExecMem level0ExecMem;
209 0 : level0ExecMem.count = param.DataDes.count * level1RankSize_;
210 0 : level0ExecMem.inputMem = DeviceMem::create(inputPtr, inputSize);
211 0 : level0ExecMem.outputMem = DeviceMem::create(outputPtr, blockSize_);
212 0 : level0ExecMem.inputPtr = inputPtr;
213 0 : level0ExecMem.outputPtr = outputPtr;
214 0 : CHK_RET(KernelRunIntraServerPost(param, level0ExecMem));
215 0 : }
216 :
217 : // L2 RDMA
218 0 : if (step < level2RankSize_ - 1) {
219 0 : blockIdx_ = (blockIdx_ + level2RankSize_ - 1) % level2RankSize_;
220 0 : CHK_RET(KernelRunInterSuperPod(param, execMem));
221 0 : memIdx_ = 1 - memIdx_;
222 0 : CHK_RET(WaitRdmaStreamFinish());
223 : }
224 : }
225 :
226 0 : return HCCL_SUCCESS;
227 0 : }
228 :
229 : /* 超节点间1步 RMDA通信 */
230 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::KernelRunInterSuperPod(const OpParam ¶m, ExecMem &execMem)
231 : {
232 : (void)param;
233 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
234 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
235 :
236 0 : u32 prevLevel2Rank = (level2Rank_ + level2RankSize_ - 1) % level2RankSize_;
237 0 : u32 nextLevel2Rank = (level2Rank_ + 1) % level2RankSize_;
238 0 : LINK prevLevel2Link = level2CommInfo.links[prevLevel2Rank];
239 0 : LINK nextLevel2Link = level2CommInfo.links[nextLevel2Rank];
240 :
241 0 : CHK_RET(prevLevel2Link->TxAck(rdmaMainStream_));
242 0 : CHK_RET(nextLevel2Link->RxAck(rdmaMainStream_));
243 :
244 0 : u64 curSize = execMem.count * unitSize_;
245 0 : CHK_PRT_RET(memIdx_ > 1, HCCL_ERROR("[KernelRunInterSuperPod]memIdx[%u] is not valid", memIdx_), HCCL_E_PARA);
246 0 : u64 srcOffset = memIdx_ * curSize; // memIdx=0时发是0，收是1；memIdx=1时发是1，收是0
247 0 : u64 dstOffset = (1 - memIdx_) * curSize;
248 0 : HCCL_INFO("[KernelRunInterSuperPod] local rank[%u] to[%u] from[%u] srcOffset[%llu] dstOffset[%llu] size[%llu]",
249 : level2Rank_, nextLevel2Rank, prevLevel2Rank, srcOffset, dstOffset, curSize);
250 :
251 0 : CHK_RET(nextLevel2Link->TxAsync(UserMemType::INPUT_MEM, dstOffset,
252 : static_cast<u8 *>(execMem.inputMem.ptr()) + srcOffset, curSize, rdmaMainStream_));
253 0 : CHK_RET(prevLevel2Link->RxAsync(UserMemType::INPUT_MEM, srcOffset,
254 : static_cast<u8 *>(execMem.inputMem.ptr()) + dstOffset, curSize, rdmaMainStream_));
255 0 : CHK_RET(prevLevel2Link->PostFinAck(rdmaMainStream_));
256 0 : CHK_RET(nextLevel2Link->WaitFinAck(rdmaMainStream_));
257 0 : return HCCL_SUCCESS;
258 0 : }
259 :
260 : /* Server内的零拷贝通信 */
261 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::KernelRunIntraServerPost(const OpParam ¶m, ExecMem &execMem)
262 : {
263 : // 计算slice
264 0 : u64 sliceSize = execMem.count * unitSize_;
265 0 : std::vector<Slice> dataSegsSlice;
266 0 : CalcDataSlices(sliceSize, level0RankSize_, dataSegsSlice);
267 :
268 : // 执行AllGather
269 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice = {dataSegsSlice};
270 0 : u64 baseOffset = blockIdx_ * blockSize_;
271 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) { // 只有1个ring环，只用到一条流，没有流之间同步
272 0 : HCCL_INFO("[%s] single ring AllGather", __func__);
273 0 : CHK_RET(MultiRingAllGather(tag_, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType,
274 : multRingsUserMemSlice, mainStream_, PROF_STAGE_0, baseOffset, nullptr, multRingsUserMemSlice));
275 : } else {
276 0 : CHK_PRT_RET(topoType_ != TopoType::TOPO_TYPE_NP_DOUBLE_RING,
277 : HCCL_ERROR("[%s] unknown topoType: %u", __func__, topoType_), HCCL_E_NOT_SUPPORT);
278 0 : HCCL_INFO("[%s] semi ring AllGather", __func__);
279 0 : CHK_RET(SemiRingAllGather(tag_, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType,
280 : multRingsUserMemSlice, mainStream_, PROF_STAGE_0, baseOffset, nullptr, multRingsUserMemSlice));
281 : }
282 :
283 0 : return HCCL_SUCCESS;
284 0 : }
285 :
286 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::SemiRingAllGather(
287 : const std::string &tag, DeviceMem &inputMem, DeviceMem &outputMem,
288 : const u64 count, const HcclDataType &dataType, const std::vector<std::vector<Slice>> &multRingsSliceZero,
289 : const Stream &stream, s32 profStage, const u64 baseOffset, const HcomCollOpInfo *opInfo,
290 : const std::vector<std::vector<Slice>> &multRingsUserMemSlice)
291 : {
292 : (void)tag;
293 : (void)multRingsSliceZero;
294 : (void)opInfo;
295 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
296 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
297 :
298 : // 执行
299 0 : std::unique_ptr<AlgTemplateBase> level0Template = AlgTemplateRegistry::Instance().GetAlgTemplate(
300 0 : TemplateType::TEMPLATE_ALL_GATHER_UNIFIED_MARCH, dispatcher_);
301 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_UNIFIED_MARCH in COMM_LEVEL0", __func__);
302 0 : CHK_SMART_PTR_NULL(level0Template);
303 :
304 0 : CHK_RET(level0Template->Prepare(stream, level0CommInfo, inputMem, outputMem,
305 : inputMem, outputMem, count * SIZE_TABLE[dataType], sdmaSubStreams_, notifySdmaMain_,
306 : notifySdmaSub_, multRingsUserMemSlice, baseOffset));
307 0 : HcclResult ret = level0Template->RegisterProfiler(
308 : ((COMM_INDEX_0 + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) +
309 0 : (level0CommInfo.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level0CommInfo.localRank,
310 : profStage, HCCL_EXEC_STEP_NOT_SET, stream);
311 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
312 : HCCL_ERROR("[CollAllGatherRingZerocopyPipelineExecutor][SemiRingAllGather]SemiRing AllGather failed, ret[%d]",
313 : ret), ret);
314 0 : CHK_RET(level0Template->RunAsync());
315 0 : return ret;
316 0 : }
317 :
318 : /* 超节点内的节点间通信 */
319 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::KernelRunInterServer(const OpParam ¶m, ExecMem &execMem)
320 : {
321 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherRingZerocopyPipelineExecutor][KernelRunInterServer] starts");
322 :
323 : // 前处理
324 0 : CHK_RET(KernelRunInterServerPreProcess(param, execMem));
325 :
326 0 : if (level1RankSize_ > 1) {
327 : // 计算slice
328 0 : u64 sliceSize = execMem.count * unitSize_;
329 0 : std::vector<Slice> level1DataSegsSlice;
330 0 : CalcDataSlices(sliceSize, level1RankSize_, level1DataSegsSlice);
331 :
332 0 : std::unique_ptr<AlgTemplateBase> level1AGTemplate;
333 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
334 0 : level1AGTemplate = AlgTemplateRegistry::Instance().GetAlgTemplate(
335 0 : TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
336 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL1", __func__);
337 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
338 0 : level1AGTemplate = AlgTemplateRegistry::Instance().GetAlgTemplate(
339 0 : TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
340 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL1", __func__);
341 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
342 0 : level1AGTemplate = AlgTemplateRegistry::Instance().GetAlgTemplate(
343 0 : TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
344 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
345 : } else {
346 0 : HCCL_ERROR("[KernelRunInterServer] unsupported level1 algtype [%s]", AlgTypeToStr(algType_).c_str());
347 0 : return HCCL_E_NOT_SUPPORT;
348 : }
349 0 : CHK_SMART_PTR_NULL(level1AGTemplate);
350 : // 执行算法编排
351 0 : CHK_RET(level1AGTemplate->Prepare(execMem.outputMem, execMem.outputMem, execMem.inputMem, INVALID_U64,
352 : param.DataDes.dataType, mainStream_, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level1DataSegsSlice));
353 0 : CHK_RET(level1AGTemplate->RegisterProfiler((level1RankSize_ << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1Rank_,
354 : PROF_STAGE_1, HCCL_EXEC_STEP_NOT_SET, mainStream_));
355 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, level0Rank_ + 1));
356 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, level0Rank_);
357 0 : CHK_RET(RunTemplate(level1AGTemplate, level1CommInfo));
358 0 : }
359 :
360 : // 后处理
361 0 : CHK_RET(KernelRunInterServerPostProcess(param, execMem));
362 :
363 0 : HCCL_INFO("[CollAllGatherRingZerocopyPipelineExecutor][KernelRunInterServer] run success");
364 0 : return HCCL_SUCCESS;
365 : }
366 :
367 : /* 数据交换 */
368 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::KernelRunInterServerPreProcess(const OpParam ¶m,
369 : ExecMem &execMem)
370 : {
371 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherRingZerocopyPipelineExecutor] KernelRunInterServerPreProcess");
372 0 : u64 curSize = execMem.count * unitSize_;
373 :
374 0 : u32 remoteRankSend = 0;
375 0 : u32 remoteRankRecv = 0;
376 0 : CHK_RET(CalExchangeRemoteRank(remoteRankSend, remoteRankRecv));
377 :
378 0 : DeviceMem srcMem = execMem.inputMem.range(memIdx_ * curSize, curSize);
379 0 : DeviceMem dstMem = execMem.outputMem.range(level1Rank_ * curSize, curSize);
380 0 : if (remoteRankSend != topoAttr_.userRank && remoteRankRecv != topoAttr_.userRank) { // 需要交换数据
381 0 : HCCL_DEBUG("[%s] rank [%u] need exchange", __func__, topoAttr_.userRank);
382 : // 获取通信对端的link
383 0 : LINK sendLink;
384 0 : LINK recvLink;
385 0 : CHK_RET(GetTransportForExchange(remoteRankSend, sendLink));
386 0 : CHK_RET(GetTransportForExchange(remoteRankRecv, recvLink));
387 :
388 0 : bool IsRemoteRankSendNeighbor = IsLevel0Neighbor(remoteRankSend, level0RankSize_);
389 0 : bool IsRemoteRankRecvNeighbor = IsLevel0Neighbor(remoteRankRecv, level0RankSize_);
390 : // 当通信对端恰好是同server邻居时，复用Level0建链，其注册内存是UserMem，需特殊处理：经过UserOut中转CCL的数据
391 : // 否则，在CommCombineOrder上建链，其注册内存是CCL Buffer
392 0 : u64 blockOffset = blockSize_ * blockIdx_;
393 0 : u64 tmpOffset = blockOffset + totalSize_ * level0Rank_ * level1RankSize_;
394 0 : DeviceMem tmpMem = DeviceMem::create(static_cast<u8 *>(execMem.outputPtr) + tmpOffset, curSize);
395 0 : if (IsRemoteRankSendNeighbor) {
396 0 : HCCL_DEBUG("[%s] neighbor process srcPtr[%p] tmpPtr[%p] size[%llu]", __func__, srcMem.ptr(), tmpMem.ptr(),
397 : curSize);
398 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, tmpMem, srcMem, mainStream_));
399 : }
400 :
401 : // 执行通信
402 0 : CHK_RET(recvLink->TxAck(mainStream_));
403 0 : CHK_RET(sendLink->RxAck(mainStream_));
404 :
405 0 : u32 remoteLevel1RankSend = remoteRankSend % (level0RankSize_ * level1RankSize_) / level0RankSize_;
406 0 : u64 txDstOffset = remoteLevel1RankSend * curSize;
407 0 : HCCL_INFO("[%s] remoteRankSend[%u] txDstOffset[%llu]", __func__, remoteRankSend, txDstOffset);
408 0 : if (IsRemoteRankSendNeighbor) {
409 0 : CHK_RET(sendLink->TxAsync(UserMemType::OUTPUT_MEM, txDstOffset, tmpMem.ptr(), curSize, mainStream_));
410 : } else {
411 0 : CHK_RET(sendLink->TxAsync(UserMemType::OUTPUT_MEM, txDstOffset, srcMem.ptr(), curSize, mainStream_));
412 : }
413 :
414 0 : u64 rxSrcOffset = memIdx_ * curSize;
415 0 : if (IsRemoteRankRecvNeighbor) {
416 0 : u32 remoteLevel0RankRecv = remoteRankRecv % level0RankSize_;
417 0 : rxSrcOffset = static_cast<u8 *>(execMem.outputPtr) - static_cast<u8 *>(param.outputPtr) +
418 0 : blockOffset + totalSize_ * remoteLevel0RankRecv * level1RankSize_;
419 0 : CHK_RET(recvLink->RxAsync(UserMemType::OUTPUT_MEM, rxSrcOffset, dstMem.ptr(), curSize, mainStream_));
420 : } else {
421 0 : CHK_RET(recvLink->RxAsync(UserMemType::INPUT_MEM, rxSrcOffset, dstMem.ptr(), curSize, mainStream_));
422 : }
423 0 : HCCL_INFO("[%s] remoteRankRecv[%u] rxSrcOffset[%llu]", __func__, remoteRankRecv, rxSrcOffset);
424 :
425 : // 交换数据的两端之间Barrier，确认收发完成
426 0 : CHK_RET(recvLink->TxAck(mainStream_));
427 0 : CHK_RET(sendLink->RxAck(mainStream_));
428 0 : CHK_RET(sendLink->TxDataSignal(mainStream_));
429 0 : CHK_RET(recvLink->RxDataSignal(mainStream_));
430 0 : } else { // 不需要交换数据，将数据从ccl in拷到ccl out
431 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, mainStream_));
432 : }
433 :
434 0 : return HCCL_SUCCESS;
435 0 : }
436 :
437 : /* 将通信结果从ccl output搬到user output */
438 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::KernelRunInterServerPostProcess(const OpParam ¶m,
439 : ExecMem &execMem)
440 : {
441 : (void)param;
442 0 : u64 blockOffset = blockIdx_ * blockSize_;
443 0 : u64 curSize = execMem.count * unitSize_;
444 0 : for (u32 i = 0; i < level1RankSize_; i++) {
445 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8 *>(execMem.outputMem.ptr()) + i * curSize, curSize);
446 0 : u64 outputOffset = blockOffset + totalSize_ * (level0Rank_ * level1RankSize_ + i);
447 0 : DeviceMem dstMem = DeviceMem::create(static_cast<u8 *>(execMem.outputPtr) + outputOffset, curSize);
448 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, mainStream_));
449 0 : HCCL_DEBUG("[%s] memcopy from CCLOut[%p] to UserOut[%p]", __func__, srcMem.ptr(), dstMem.ptr());
450 0 : }
451 0 : return HCCL_SUCCESS;
452 : }
453 :
454 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::NotifyRdmaStreamStart()
455 : {
456 0 : CHK_RET(LocalNotify::Post(mainStream_, dispatcher_, notifyMainToRdma_, INVALID_VALUE_STAGE));
457 0 : CHK_RET(LocalNotify::Wait(rdmaMainStream_, dispatcher_, notifyMainToRdma_, INVALID_VALUE_STAGE));
458 0 : return HCCL_SUCCESS;
459 : }
460 :
461 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::WaitRdmaStreamFinish()
462 : {
463 0 : CHK_RET(LocalNotify::Post(rdmaMainStream_, dispatcher_, notifyRdmaToMain_, INVALID_VALUE_STAGE));
464 0 : CHK_RET(LocalNotify::Wait(mainStream_, dispatcher_, notifyRdmaToMain_, INVALID_VALUE_STAGE));
465 0 : return HCCL_SUCCESS;
466 : }
467 :
468 : /* 建链时也需要调用，还没有commInfo */
469 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalExchangeRemoteRank(u32 &remoteRankSend, u32 &remoteRankRecv)
470 : {
471 0 : u32 userRank = topoAttr_.userRank;
472 0 : u32 userRankSize = topoAttr_.userRankSize;
473 0 : u32 level2RankSize = topoAttr_.superPodNum;
474 0 : CHK_PRT_RET(level2RankSize == 0, HCCL_ERROR("[CalExchangeRemoteRank]level2RankSize is 0"), HCCL_E_PARA);
475 0 : u32 level1RankSize = topoAttr_.serverNum / level2RankSize;
476 0 : CHK_PRT_RET(level1RankSize == 0, HCCL_ERROR("[CalExchangeRemoteRank]level1RankSize is 0"), HCCL_E_PARA);
477 0 : u32 level0RankSize = userRankSize / level1RankSize / level2RankSize;
478 0 : u32 level0Rank = userRank % level0RankSize;
479 0 : u32 level1Rank = userRank % (level0RankSize * level1RankSize) / level0RankSize;
480 0 : u32 level2Rank = userRank / level0RankSize / level1RankSize;
481 :
482 0 : u32 level2StartRank = level2Rank * level0RankSize * level1RankSize;
483 : // 计算本超节点内本端将要接收数据的源rank
484 0 : remoteRankRecv = level2StartRank + level0Rank * level1RankSize + level1Rank;
485 : // 计算本超节点内本端将要发送数据的目标rank
486 0 : u32 srcLevel0Rank = (userRank - level2StartRank) / level1RankSize;
487 0 : u32 srcLevel1Rank = (userRank - level2StartRank) % level1RankSize;
488 0 : remoteRankSend = level2StartRank + srcLevel1Rank * level0RankSize + srcLevel0Rank;
489 :
490 0 : HCCL_INFO("[%s] rank[%u:%u/%u/%u] remoteRankSend[%u], remoteRankRecv[%u]", __func__, topoAttr_.userRank, level2Rank,
491 : level1Rank, level0Rank, remoteRankSend, remoteRankRecv);
492 0 : return HCCL_SUCCESS;
493 : }
494 :
495 0 : HcclResult CollAllGatherRingZerocopyPipelineExecutor::CalcDataSlices(u64 sliceSize, u32 rankSize,
496 : std::vector<Slice> &dataSegsSlice)
497 : {
498 0 : dataSegsSlice.resize(rankSize);
499 0 : for (u32 i = 0; i < rankSize; i++) {
500 0 : dataSegsSlice[i].size = sliceSize;
501 0 : dataSegsSlice[i].offset = i * sliceSize;
502 : }
503 0 : return HCCL_SUCCESS;
504 : }
505 :
506 : REGISTER_EXEC("AllGatherRingZerocopyPipelineExecutor", AllGatherRingZerocopyPipeline,
507 : CollAllGatherRingZerocopyPipelineExecutor);
508 : } // namespace hccl
|
