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