Line data Source code
1 : /**
2 : * Copyright (c) 2026 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_pipeline_for_910_93_executor.h"
12 : #include "hccl_types.h"
13 : #include "alg_template_register.h"
14 : #include "alg_template_base_pub.h"
15 :
16 : namespace hccl {
17 : constexpr u32 PIPELINE_NUM = 2;
18 : constexpr u32 PIPELINE_EXTRA_STREAM_NUM = PIPELINE_NUM;
19 :
20 0 : CollAllGatherPipelineFor91093Executor::CollAllGatherPipelineFor91093Executor(
21 0 : const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
22 0 : : CollAllGatherExecutor(dispatcher, topoMatcher)
23 : {
24 0 : DMAReduceFlag_ = workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE;
25 : desc_.level1SupportedAlgos
26 0 : = {AlgTypeLevel1::ALG_LEVEL1_NHR, AlgTypeLevel1::ALG_LEVEL1_NB, AlgTypeLevel1::ALG_LEVEL1_RING};
27 : desc_.level2SupportedAlgos
28 0 : = {AlgTypeLevel2::ALG_LEVEL2_NHR, AlgTypeLevel2::ALG_LEVEL2_NB, AlgTypeLevel2::ALG_LEVEL2_RING};
29 0 : }
30 :
31 0 : HcclResult CollAllGatherPipelineFor91093Executor::CalcStreamNum(u32& streamNum)
32 : {
33 : // 计算三级流水线所需的流数量
34 0 : HCCL_INFO(
35 : "[CollAllGatherPipelineFor91093Executor][CalcStreamNum] topoType_[%u], workflowMode_[%u]", topoType_,
36 : workflowMode_);
37 : // 基本流数量计算
38 0 : u32 totalStreamNum
39 0 : = (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING ? LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE :
40 : LEVEL0_PLANE_NUM_IN_NPRING_SINGLE);
41 0 : if (workflowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) { // 工作流模式，双倍的流，用于并行操作
42 0 : totalStreamNum *= STREAM_NUM_FOR_DMAREDUCE_ONE_RING; // *2
43 : }
44 :
45 : // 为三级流水线增加额外的流
46 : // 从流用于L2，主流用于L1 + L0
47 : // 新增从流中，一条用于L2流水线，一条用于多申请2个notify，共新增4个notify用于两块内存的主从流之间的同步
48 0 : totalStreamNum += PIPELINE_EXTRA_STREAM_NUM;
49 :
50 0 : streamNum = totalStreamNum - 1;
51 0 : HCCL_INFO("[CollAllGatherPipelineFor91093Executor][CalcStreamNum] tag[%s] streamNum[%u]", tag_.c_str(), streamNum);
52 0 : return HCCL_SUCCESS;
53 : }
54 :
55 0 : HcclResult CollAllGatherPipelineFor91093Executor::CalcCommInfo(std::vector<LevelNSubCommTransport>& opTransport)
56 : {
57 0 : TransportMemType inputType = TransportMemType::RESERVED;
58 0 : TransportMemType outputType = TransportMemType::RESERVED;
59 0 : CHK_RET(CalcTransportMemType(inputType, outputType));
60 0 : CHK_RET(CalcLevel0CommInfo(inputType, outputType, opTransport));
61 0 : CHK_RET(CalcLevel1CommInfo(inputType, outputType, opTransport));
62 0 : CHK_RET(CalcLevel2CommInfo(inputType, outputType, opTransport));
63 0 : return HCCL_SUCCESS;
64 : }
65 :
66 : // level0 ring
67 0 : HcclResult CollAllGatherPipelineFor91093Executor::CalcLevel0CommInfo(
68 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
69 : {
70 0 : CommParaInfo commParaInfo(COMM_LEVEL0, CommType::COMM_TAG_RING_INNER);
71 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL0], inputType, outputType));
72 0 : return HCCL_SUCCESS;
73 0 : }
74 :
75 : // level2 NHR
76 0 : HcclResult CollAllGatherPipelineFor91093Executor::CalcLevel2CommInfo(
77 : TransportMemType inputType, TransportMemType outputType, std::vector<LevelNSubCommTransport>& opTransport)
78 : {
79 0 : CommParaInfo commParaInfo(COMM_LEVEL2, CommType::COMM_TAG_MAX);
80 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
81 0 : commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING;
82 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
83 0 : commParaInfo.commType = CommType::COMM_TAG_NONUNIFORM_BRUCK;
84 : } else {
85 0 : commParaInfo.commType = CommType::COMM_TAG_RING_INNER;
86 : }
87 0 : CHK_RET(CalcCommPlaneInfo(tag_, commParaInfo, opTransport[COMM_LEVEL2], inputType, outputType));
88 0 : return HCCL_SUCCESS;
89 0 : }
90 :
91 : HcclResult
92 0 : CollAllGatherPipelineFor91093Executor::CalcTransportMemType(TransportMemType& inputType, TransportMemType& outputType)
93 : {
94 0 : inputType = TransportMemType::CCL_INPUT;
95 0 : outputType = TransportMemType::CCL_OUTPUT;
96 0 : HCCL_INFO(
97 : "[CollAllGatherPipelineFor91093Executor][CalcTransportMemType]"
98 : "tag[%s] inputType[%d], outputType[%d]",
99 : tag_.c_str(), inputType, outputType);
100 0 : return HCCL_SUCCESS;
101 : }
102 :
103 : // 每次循环处理的数据量，双流水的情况下需要满足每个流水线都能满载
104 0 : u64 CollAllGatherPipelineFor91093Executor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
105 : {
106 : // 分成两片，做流水ping-pong
107 0 : u64 maxCountPerLoop
108 0 : = cclBuffSize / PIPELINE_NUM / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
109 0 : HCCL_INFO("[%s] tag[%s] maxCountPerLoop[%llu]", __func__, tag_.c_str(), maxCountPerLoop);
110 :
111 0 : return maxCountPerLoop;
112 : }
113 :
114 : // 编排
115 0 : HcclResult CollAllGatherPipelineFor91093Executor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
116 : {
117 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[CollAllGatherPipelineFor91093Executor][Orchestrate] begins.");
118 :
119 0 : HcclUs startut = TIME_NOW();
120 0 : tag_ = param.tag;
121 0 : algResResp_ = &algRes;
122 :
123 : // 设置L0和L1通信域信息
124 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, COMM_INDEX_0 + 1));
125 0 : level0CommInfo_ = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
126 0 : u32 commIndex = level0CommInfo_.localRank;
127 0 : CHK_RET(CheckCommSize(COMM_LEVEL1, commIndex + 1));
128 0 : level1CommInfo_ = GetSubCommInfo(COMM_LEVEL1, commIndex);
129 : // 获取L2通信域信息
130 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0 + 1));
131 0 : level2CommInfo_ = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
132 :
133 : // L1/L0 runs on the main stream; L2 runs on the reserved slave stream.
134 0 : mainStreamL1L0_ = param.stream;
135 0 : subStreams_ = algResResp_->slaveStreams;
136 0 : mainStreamL2_ = subStreams_.back();
137 0 : const u32 baseStreamNum = algResResp_->slaveStreams.size() - PIPELINE_EXTRA_STREAM_NUM;
138 0 : notifyL1L0ToL2A_ = algResResp_->notifiesAux[baseStreamNum];
139 0 : notifyL1L0ToL2B_ = algResResp_->notifiesAux[baseStreamNum + 1];
140 0 : notifyL2ToL1L0A_ = algResResp_->notifiesMain[baseStreamNum];
141 0 : notifyL2ToL1L0B_ = algResResp_->notifiesMain[baseStreamNum + 1];
142 0 : HCCL_INFO(
143 : "[CollAllGatherPipelineFor91093Executor][RunLoop] NotifyIds: "
144 : "L1L0ToL2A: Aux[%u], L1L0ToL2B: Aux[%u], L2ToL1L0A: Main[%u], L2ToL1L0B: Main[%u]",
145 : baseStreamNum, baseStreamNum + 1, baseStreamNum, baseStreamNum + 1);
146 0 : notifyRingMain_.assign(
147 0 : algResResp_->notifiesMain.begin(), algResResp_->notifiesMain.end() - PIPELINE_EXTRA_STREAM_NUM);
148 0 : notifyRingSub_.assign(algResResp_->notifiesAux.begin(), algResResp_->notifiesAux.end() - PIPELINE_EXTRA_STREAM_NUM);
149 0 : ringSubStreams_.assign(subStreams_.begin(), subStreams_.end() - PIPELINE_EXTRA_STREAM_NUM);
150 :
151 : // 计算通信域信息和内存类型
152 0 : unitSize_ = SIZE_TABLE[param.DataDes.dataType];
153 0 : cclInputSizeHalved_ = algResResp_->cclInputMem.size() / PIPELINE_NUM;
154 0 : cclInputAMem_ = algResResp_->cclInputMem.range(0, cclInputSizeHalved_);
155 0 : cclInputBMem_ = algResResp_->cclInputMem.range(cclInputSizeHalved_, cclInputSizeHalved_);
156 0 : cclOutputSizeHalved_ = algResResp_->cclOutputMem.size() / PIPELINE_NUM;
157 0 : cclOutputAMem_ = algResResp_->cclOutputMem.range(0, cclOutputSizeHalved_);
158 0 : cclOutputBMem_ = algResResp_->cclOutputMem.range(cclOutputSizeHalved_, cclOutputSizeHalved_);
159 :
160 0 : CHK_RET(RunLoop(param)); // 运行循环，循环内执行三级流水线
161 :
162 0 : HCCL_INFO(
163 : "tag[%s], Allgather executor orchestrate success, take time [%lld]us.", tag_.c_str(),
164 : DURATION_US(TIME_NOW() - startut));
165 :
166 0 : return HCCL_SUCCESS;
167 : }
168 :
169 0 : HcclResult CollAllGatherPipelineFor91093Executor::RunL2Stage(
170 : const OpParam& param, ExecMem& execMem, u64 loopIdx, u64 memIdx, u64 bufferSliceNum)
171 : {
172 : // superpod数量不超过1不需要跨超节点；最后一轮循环处理L1L0的最后一片数据，L2不需要参与通信，跳过L2阶段
173 0 : if (loopIdx >= bufferSliceNum) {
174 0 : return HCCL_SUCCESS;
175 : }
176 : // Loop 0 waits for the main stream start signal. Later ping-pong buffer reuse waits for L1/L0.
177 0 : if (loopIdx == 0 || loopIdx >= PIPELINE_NUM) {
178 0 : auto notifyL1L0ToL2 = (memIdx == 0) ? notifyL1L0ToL2A_ : notifyL1L0ToL2B_;
179 0 : CHK_RET(LocalNotify::Wait(mainStreamL2_, dispatcher_, notifyL1L0ToL2, INVALID_VALUE_STAGE));
180 0 : }
181 :
182 : // Local Copy: UserIn -> Ccl
183 0 : u64 curSize = execMem.count * unitSize_;
184 0 : DeviceMem srcMem = DeviceMem::create(static_cast<u8*>(execMem.inputPtr), curSize);
185 0 : DeviceMem dstMem = execMem.inputMem.range(0, curSize);
186 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, mainStreamL2_));
187 :
188 : // OPBASE模式
189 0 : u64 dstMemOffset = topoAttr_.userRank * curSize;
190 0 : DeviceMem dmaDst = execMem.outputMem.range(dstMemOffset, curSize);
191 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dmaDst, srcMem, mainStreamL2_));
192 :
193 : // L2 Rx -> DMA[0/1]
194 0 : u64 baseOffset = memIdx == 0 ? 0 : cclInputSizeHalved_;
195 0 : CHK_RET(KernelRunInterSuperPod(param, execMem, baseOffset));
196 0 : auto notifyL2ToL1L0 = (memIdx == 0) ? notifyL2ToL1L0A_ : notifyL2ToL1L0B_;
197 0 : CHK_RET(LocalNotify::Post(mainStreamL2_, dispatcher_, notifyL2ToL1L0, INVALID_VALUE_STAGE));
198 0 : return HCCL_SUCCESS;
199 0 : }
200 :
201 0 : HcclResult CollAllGatherPipelineFor91093Executor::RunL1L0Stage(
202 : const OpParam& param, ExecMem& lastExecMem, u64 loopIdx, u64 memIdx, u64 bufferSliceNum)
203 : {
204 : // 第一轮等待L2处理完
205 0 : if (loopIdx < 1) {
206 0 : return HCCL_SUCCESS;
207 : }
208 : // 同步：等待上一轮的 L2 通信完成
209 0 : auto notifyL2ToL1L0 = (memIdx == 0) ? notifyL2ToL1L0A_ : notifyL2ToL1L0B_;
210 0 : CHK_RET(LocalNotify::Wait(mainStreamL1L0_, dispatcher_, notifyL2ToL1L0, INVALID_VALUE_STAGE));
211 :
212 0 : u64 baseOffset = memIdx == 0 ? 0 : cclInputSizeHalved_;
213 0 : if (level1CommInfo_.localRankSize > 1) {
214 0 : CHK_RET(KernelRunInterServer(param, lastExecMem, baseOffset));
215 : }
216 0 : CHK_RET(KernelRunIntraServer(param, lastExecMem, baseOffset));
217 :
218 0 : if (loopIdx + 1 < bufferSliceNum) {
219 0 : auto notifyL1L0ToL2 = (memIdx == 0) ? notifyL1L0ToL2A_ : notifyL1L0ToL2B_;
220 0 : CHK_RET(LocalNotify::Post(mainStreamL1L0_, dispatcher_, notifyL1L0ToL2, INVALID_VALUE_STAGE));
221 0 : }
222 0 : return HCCL_SUCCESS;
223 0 : }
224 :
225 0 : HcclResult CollAllGatherPipelineFor91093Executor::RunLoop(OpParam& param)
226 : {
227 0 : u8* userInputPtr = static_cast<u8*>(param.inputPtr);
228 0 : u8* userOutputPtr = static_cast<u8*>(param.outputPtr);
229 0 : CHK_PTR_NULL(userInputPtr);
230 0 : CHK_PTR_NULL(userOutputPtr);
231 :
232 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algResResp_->cclInputMem.size(), unitSize_);
233 0 : CHK_PRT_RET(
234 : maxCountPerLoop == 0,
235 : HCCL_ERROR(
236 : "[CollAllGatherPipelineFor91093Executor][RunLoop]tag[%s] userRankSize[%u] maxCountPerLoop[%llu]",
237 : tag_.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
238 : HCCL_E_PARA);
239 0 : u64 bufferSliceNum = (param.DataDes.count + maxCountPerLoop - 1) / maxCountPerLoop;
240 0 : if (bufferSliceNum == 0) {
241 0 : return HCCL_SUCCESS;
242 : }
243 0 : HCCL_INFO(
244 : "[CollAllGatherPipelineFor91093Executor][%s] maxCountPerLoop[%llu] bufferSliceNum[%llu]", __func__,
245 : maxCountPerLoop, bufferSliceNum);
246 0 : u64 loopNum = bufferSliceNum + 1;
247 0 : u64 countLeft = param.DataDes.count; // 剩余的数据量
248 :
249 0 : u32 memIdx = 0;
250 0 : ExecMem lastExecMem;
251 : // AllGather starts with L2, so the main stream releases the first L2 stage before the loop.
252 0 : CHK_RET(LocalNotify::Post(mainStreamL1L0_, dispatcher_, notifyL1L0ToL2A_, INVALID_VALUE_STAGE));
253 0 : for (u64 loopIdx = 0; loopIdx < loopNum; loopIdx++) {
254 0 : u64 curCount = countLeft > maxCountPerLoop ? maxCountPerLoop : countLeft; // 当前循环处理的数据量
255 0 : countLeft -= curCount;
256 :
257 0 : ExecMem execMem;
258 0 : execMem.count = curCount;
259 0 : execMem.inputMem = memIdx == 0 ? cclInputAMem_ : cclInputBMem_;
260 0 : execMem.outputMem = memIdx == 0 ? cclOutputAMem_ : cclOutputBMem_;
261 0 : execMem.inputPtr = userInputPtr;
262 0 : execMem.outputPtr = userOutputPtr;
263 :
264 0 : CHK_RET(RunL2Stage(param, execMem, loopIdx, memIdx, bufferSliceNum));
265 0 : CHK_RET(RunL1L0Stage(param, lastExecMem, loopIdx, 1 - memIdx, bufferSliceNum));
266 :
267 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
268 :
269 0 : u64 curSize = curCount * unitSize_;
270 0 : userInputPtr += curSize;
271 0 : userOutputPtr += curSize;
272 0 : memIdx = 1 - memIdx; // 双缓冲交替使用
273 0 : lastExecMem = execMem;
274 0 : }
275 :
276 0 : return HCCL_SUCCESS;
277 0 : }
278 :
279 : // 跨超节点
280 : HcclResult
281 0 : CollAllGatherPipelineFor91093Executor::KernelRunInterSuperPod(const OpParam& param, ExecMem& execMem, u64 baseOffset)
282 : {
283 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] begins, topoType_[%u], DMAReduceFlag_[%u]", __func__, topoType_, DMAReduceFlag_);
284 0 : std::unique_ptr<AlgTemplateBase> level2AGExecutor;
285 0 : if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NB) {
286 : level2AGExecutor
287 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
288 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL2", __func__);
289 0 : } else if (algType_.algoLevel2 == AlgTypeLevel2::ALG_LEVEL2_NHR) {
290 : level2AGExecutor
291 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
292 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL2", __func__);
293 : } else {
294 : level2AGExecutor
295 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
296 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL2", __func__);
297 : }
298 0 : CHK_SMART_PTR_NULL(level2AGExecutor);
299 :
300 0 : u64 curDataSegsSliceSize = execMem.count * unitSize_;
301 : std::vector<Slice> level2DataSegsSlice
302 0 : = PrepareSlicesL2(param, level2CommInfo_, level1CommInfo_, level0CommInfo_, unitSize_, curDataSegsSliceSize);
303 0 : CHK_RET(level2AGExecutor->Prepare(
304 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, mainStreamL2_,
305 : HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level2DataSegsSlice, baseOffset));
306 :
307 0 : CHK_RET(level2AGExecutor->RegisterProfiler(
308 : (level2CommInfo_.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo_.localRank, PROF_STAGE_0,
309 : HCCL_EXEC_STEP_NOT_SET, mainStreamL2_));
310 :
311 0 : CHK_RET(RunTemplate(level2AGExecutor, level2CommInfo_));
312 0 : HCCL_INFO("[%s] AllGather level2 AllGather run success, topoType_[%u]", __func__, topoType_);
313 0 : return HCCL_SUCCESS;
314 0 : }
315 :
316 : // 超节点内的节点间通信 L1nhr
317 : HcclResult
318 0 : CollAllGatherPipelineFor91093Executor::KernelRunInterServer(const OpParam& param, ExecMem& execMem, u64 baseOffset)
319 : {
320 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] begins, topoType_[%u], DMAReduceFlag_[%u]", __func__, topoType_, DMAReduceFlag_);
321 0 : u64 curDataSegsSliceSize = execMem.count * unitSize_;
322 : std::vector<Slice> level1DataSegsSlice
323 0 : = PrepareSlicesL1(param, level2CommInfo_, level1CommInfo_, level0CommInfo_, unitSize_, curDataSegsSliceSize);
324 :
325 0 : std::unique_ptr<AlgTemplateBase> level1AGExecutor;
326 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
327 : level1AGExecutor
328 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
329 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_RING in COMM_LEVEL1", __func__);
330 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
331 : level1AGExecutor
332 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
333 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NB in COMM_LEVEL1", __func__);
334 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
335 : level1AGExecutor
336 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
337 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALL_GATHER_NHR in COMM_LEVEL1", __func__);
338 : } else {
339 0 : HCCL_ERROR("AllGather ring: unsupported algtype [%s].", AlgTypeToStr(algType_).c_str());
340 0 : return HCCL_E_NOT_SUPPORT;
341 : }
342 0 : CHK_SMART_PTR_NULL(level1AGExecutor);
343 0 : CHK_RET(level1AGExecutor->Prepare(
344 : execMem.outputMem, execMem.outputMem, execMem.inputMem, execMem.count, param.DataDes.dataType, mainStreamL1L0_,
345 : HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID, level1DataSegsSlice, baseOffset));
346 :
347 0 : CHK_RET(level1AGExecutor->RegisterProfiler(
348 : (level1CommInfo_.localRankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level1CommInfo_.localRank, PROF_STAGE_1,
349 : HCCL_EXEC_STEP_NOT_SET, mainStreamL1L0_));
350 :
351 0 : CHK_RET(RunTemplate(level1AGExecutor, level1CommInfo_));
352 0 : HCCL_INFO("[%s] AllGather level1 AllGather run success, topoType_[%u]", __func__, topoType_);
353 0 : return HCCL_SUCCESS;
354 0 : }
355 :
356 : // Server内的通信
357 : HcclResult
358 0 : CollAllGatherPipelineFor91093Executor::KernelRunIntraServer(const OpParam& param, ExecMem& execMem, u64 baseOffset)
359 : {
360 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] begins, topoType_[%u], DMAReduceFlag_[%u]", __func__, topoType_, DMAReduceFlag_);
361 : // 节点内做AllGather ring
362 0 : u64 curDataSegsSliceSize = execMem.count * unitSize_;
363 0 : std::vector<std::vector<Slice>> multRingsSlice;
364 0 : CHK_RET(PrepareSlicesL0(
365 : multRingsSlice, param, level2CommInfo_, level1CommInfo_, level0CommInfo_, unitSize_, curDataSegsSliceSize));
366 :
367 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
368 0 : CHK_RET(PrepareUserMemSlices(
369 : multRingsUserMemSlice, multRingsSlice, param, level2CommInfo_, level1CommInfo_, level0CommInfo_, unitSize_,
370 : curDataSegsSliceSize));
371 :
372 : // allgather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
373 0 : l0OpInfo_.inputAddr = nullptr;
374 0 : l0OpInfo_.outputAddr = execMem.outputPtr;
375 0 : l0OpInfo_.dataType = param.GetDataType();
376 0 : l0OpInfo_.count = execMem.count;
377 0 : l0OpInfo_.root = 0;
378 0 : l0OpInfo_.reduceOp = HCCL_REDUCE_RESERVED;
379 0 : l0OpInfo_.strideCount = param.DataDes.strideCount;
380 :
381 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
382 0 : CHK_RET(DoubleRingAllGather(
383 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, multRingsSlice,
384 : mainStreamL1L0_, PROF_STAGE_2, baseOffset, &l0OpInfo_, multRingsUserMemSlice));
385 0 : } else if (topoType_ == TopoType::TOPO_TYPE_NP_SINGLE_RING) {
386 0 : CHK_RET(MultiRingAllGather(
387 : param.tag, execMem.inputMem, execMem.outputMem, execMem.count, param.DataDes.dataType, multRingsSlice,
388 : mainStreamL1L0_, PROF_STAGE_2, baseOffset, &l0OpInfo_, multRingsUserMemSlice, COMM_LEVEL0));
389 : } else {
390 0 : return HCCL_E_NOT_SUPPORT;
391 : }
392 0 : HCCL_INFO("[%s] AllGather level0 Ring run success, topoType_[%u]", __func__, topoType_);
393 0 : return HCCL_SUCCESS;
394 0 : }
395 :
396 0 : std::vector<Slice> CollAllGatherPipelineFor91093Executor::PrepareSlicesL1(
397 : [[maybe_unused]] const OpParam& param, const SubCommInfo& level2CommInfo, const SubCommInfo& level1CommInfo,
398 : const SubCommInfo& level0CommInfo, [[maybe_unused]] u32 perDataSize, u64 inputMemSize) const
399 : {
400 0 : const u32 level0RankSize = level0CommInfo.localRankSize;
401 0 : const u32 level0ServerIndex = level0CommInfo.localRank;
402 0 : const u32 level1RankSize = level1CommInfo.localRankSize;
403 0 : const u32 level2RankSize = level2CommInfo.localRankSize;
404 0 : std::vector<Slice> level1DataSegsSlice;
405 0 : for (u32 j = 0; j < level1RankSize; j++) {
406 0 : for (u32 i = 0; i < level2RankSize; i++) {
407 0 : Slice level1Slice;
408 0 : level1Slice.size = inputMemSize;
409 : level1Slice.offset
410 0 : = inputMemSize * (i * level1RankSize * level0RankSize + j * level0RankSize + level0ServerIndex);
411 :
412 0 : HCCL_DEBUG(
413 : "[CollAllGatherPipelineFor91093Executor][PrepareSlicesL1] rank[%u], level1index[%u], level2index[%u], "
414 : "slices.offset=%llu, slices.size=%llu",
415 : level0CommInfo.localRank, j, i, level1Slice.offset, level1Slice.size);
416 :
417 0 : level1DataSegsSlice.push_back(level1Slice);
418 : }
419 : }
420 0 : return level1DataSegsSlice;
421 0 : }
422 :
423 0 : std::vector<Slice> CollAllGatherPipelineFor91093Executor::PrepareSlicesL2(
424 : [[maybe_unused]] const OpParam& param, const SubCommInfo& level2CommInfo, const SubCommInfo& level1CommInfo,
425 : const SubCommInfo& level0CommInfo, [[maybe_unused]] u32 perDataSize, u64 inputMemSize) const
426 : {
427 0 : const u32 level0RankSize = level0CommInfo.localRankSize;
428 0 : const u32 level0ServerIndex = level0CommInfo.localRank;
429 0 : const u32 level1RankSize = level1CommInfo.localRankSize;
430 0 : const u32 level1ServerIndex = level1CommInfo.localRank;
431 0 : const u32 level2RankSize = level2CommInfo.localRankSize;
432 0 : std::vector<Slice> level2DataSegsSlice;
433 0 : for (u32 i = 0; i < level2RankSize; i++) {
434 0 : Slice sliceTemp;
435 0 : sliceTemp.size = inputMemSize;
436 : sliceTemp.offset
437 0 : = inputMemSize
438 0 : * (i * level1RankSize * level0RankSize + level1ServerIndex * level0RankSize + level0ServerIndex);
439 0 : level2DataSegsSlice.push_back(sliceTemp);
440 : }
441 0 : return level2DataSegsSlice;
442 0 : }
443 :
444 0 : HcclResult CollAllGatherPipelineFor91093Executor::PrepareSlicesL0(
445 : std::vector<std::vector<Slice>>& multRingsSlice, const OpParam& param, const SubCommInfo& level2CommInfo,
446 : const SubCommInfo& level1CommInfo, const SubCommInfo& level0CommInfo, [[maybe_unused]] u32 perDataSize,
447 : u64 inputMemSize)
448 : {
449 0 : const u32 level0RankSize = level0CommInfo.localRankSize;
450 0 : const u32 level1RankSize = level1CommInfo.localRankSize;
451 0 : const u32 level2RankSize = level2CommInfo.localRankSize;
452 :
453 0 : std::vector<Slice> dataSegsSlice;
454 0 : CHK_RET(PrepareAllgatherSlice(level0RankSize, inputMemSize, dataSegsSlice));
455 :
456 : // 多环数据切分
457 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
458 :
459 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING
460 0 : && !IsSupportUnifiedMarch(param, topoType_, topoAttr_.serverNum, topoAttr_.superPodNum)) {
461 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, param.tag, false, topoAttr_.nicList);
462 : } else {
463 0 : multRingsSliceZero.push_back(dataSegsSlice);
464 : }
465 0 : for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
466 0 : std::vector<Slice> level2DataSlice;
467 0 : CHK_RET(CalculateLevel2AllgatherSlice(
468 : inputMemSize, level0RankSize, level1RankSize, level2RankSize, multRingsSliceZero, level2DataSlice,
469 : ringIndex));
470 0 : multRingsSlice.push_back(level2DataSlice);
471 0 : }
472 :
473 0 : return HCCL_SUCCESS;
474 0 : }
475 :
476 0 : HcclResult CollAllGatherPipelineFor91093Executor::PrepareUserMemSlices(
477 : std::vector<std::vector<Slice>>& userMemSlices, const std::vector<std::vector<Slice>>& multRingsSlice,
478 : const OpParam& param, [[maybe_unused]] const SubCommInfo& level2CommInfo,
479 : [[maybe_unused]] const SubCommInfo& level1CommInfo, [[maybe_unused]] const SubCommInfo& level0CommInfo,
480 : u32 perDataSize, u64 inputMemSize)
481 : {
482 0 : CHK_PRT_RET(
483 : 0 < param.DataDes.strideCount && param.DataDes.strideCount < param.DataDes.count,
484 : HCCL_ERROR(
485 : "[CollAllGatherPipelineFor91093Executor][KernelRun]strideCount[%llu] is smaller than opCount[%llu]",
486 : param.DataDes.strideCount, param.DataDes.count),
487 : HCCL_E_PARA);
488 0 : HCCL_DEBUG(
489 : "[CollAllGatherPipelineFor91093Executor][KernelRun]strideCount[%llu], opCount[%llu]", param.DataDes.strideCount,
490 : param.DataDes.count);
491 :
492 0 : for (u32 ringIndex = 0; ringIndex < multRingsSlice.size(); ringIndex++) {
493 0 : std::vector<Slice> userMemSlice;
494 0 : for (const auto& cclSlice : multRingsSlice[ringIndex]) {
495 0 : Slice tmpSlice;
496 0 : u64 count = (param.DataDes.strideCount == 0) ? param.DataDes.count : param.DataDes.strideCount;
497 0 : tmpSlice.size = cclSlice.size;
498 : tmpSlice.offset
499 0 : = (cclSlice.offset / inputMemSize) * count * perDataSize + multRingsSlice[ringIndex][0].offset;
500 0 : userMemSlice.push_back(tmpSlice);
501 0 : HCCL_DEBUG(
502 : "rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]", topoAttr_.userRank, ringIndex,
503 : tmpSlice.offset, tmpSlice.size);
504 : }
505 0 : userMemSlices.push_back(userMemSlice);
506 0 : }
507 0 : return HCCL_SUCCESS;
508 : }
509 :
510 0 : HcclResult CollAllGatherPipelineFor91093Executor::DoubleRingAllGather(
511 : const std::string& tag, DeviceMem inputMem, DeviceMem outputMem, const u64 count, const HcclDataType dataType,
512 : const std::vector<std::vector<Slice>> multRingsSliceZero, Stream stream, s32 profStage, const u64 baseOffset,
513 : const HcomCollOpInfo* opInfo, const std::vector<std::vector<Slice>> multRingsUserMemSlice)
514 : {
515 0 : HCCL_CONFIG_INFO(
516 : HCCL_ALG, "[CollAllGatherPipelineFor91093Executor]userRank[%u], count[%llu]", topoAttr_.userRank, count);
517 :
518 : (void)tag;
519 0 : HCCL_INFO("[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather] DoubleRingAllGather starts");
520 0 : HcclResult ret = HCCL_SUCCESS;
521 0 : u32 ringNum = multRingsSliceZero.size();
522 0 : CHK_RET(CheckCommSize(COMM_LEVEL0, ringNum));
523 : // 拿到ring环映射关系
524 0 : SubCommInfo level0ZeroCommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
525 0 : auto nicList = topoAttr_.nicList;
526 : std::vector<std::vector<u32>> multiRingsOrder
527 0 : = GetRingsOrderByTopoType(level0ZeroCommInfo.localRankSize, topoType_, nicList);
528 : // 生成两个ring上的userMemOut_上对应的slices
529 0 : std::vector<std::vector<Slice>> userMemOutputSlicesOfDoubleRing;
530 0 : CHK_RET(CollectMultiRingsUserMemSlices(
531 : ringNum, dataType, opInfo, multRingsSliceZero, multiRingsOrder, multRingsUserMemSlice,
532 : userMemOutputSlicesOfDoubleRing));
533 : // 生成两个ring上的rankOrder
534 0 : std::vector<std::vector<u32>> rankOrders;
535 0 : CHK_RET(CollectMultiRingsRankOrder(ringNum, multiRingsOrder, rankOrders));
536 : // 初始化executor
537 0 : std::unique_ptr<AlgTemplateBase> tempAlg = AlgTemplateRegistry::Instance().GetAlgTemplate(
538 0 : TemplateType::TEMPLATE_ALIGNED_ALL_GATHER_DOUBLE_RING, dispatcher_);
539 0 : HCCL_CONFIG_INFO(HCCL_ALG, "[%s] Run TEMPLATE_ALIGNED_ALL_GATHER_DOUBLE_RING in COMM_LEVEL0", __func__);
540 0 : CHK_SMART_PTR_NULL(tempAlg);
541 0 : CHK_RET(tempAlg->Prepare(
542 : const_cast<HcomCollOpInfo*>(opInfo), topoAttr_.userRank, ringSubStreams_, notifyRingMain_, notifyRingSub_,
543 : rankOrders, userMemOutputSlicesOfDoubleRing));
544 :
545 0 : ret = tempAlg->Prepare(
546 : outputMem, outputMem, inputMem, count, dataType, stream, multRingsSliceZero, HCCL_REDUCE_RESERVED,
547 : LEVEL0_BRIDGE_RANK_ID, baseOffset);
548 0 : CHK_PRT_RET(
549 : ret != HCCL_SUCCESS,
550 : HCCL_ERROR(
551 : "[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather]Double ring "
552 : "AllGather failed, return[%d]",
553 : ret),
554 : ret);
555 0 : u32 ringIndexOp = COMM_INDEX_0;
556 0 : u32 rankSize = level0ZeroCommInfo.localRankSize;
557 0 : ret = tempAlg->RegisterProfiler(
558 0 : ((ringIndexOp + 1) << PROF_RINGINDEX_OFFSET_OF_PLANEID) + (rankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID)
559 0 : + level0ZeroCommInfo.localRank,
560 : profStage, HCCL_EXEC_STEP_NOT_SET, stream);
561 0 : CHK_PRT_RET(
562 : ret != HCCL_SUCCESS,
563 : HCCL_ERROR(
564 : "[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather]Double ring "
565 : "AllGather failed, return[%d]",
566 : ret),
567 : ret);
568 :
569 : // 空拷贝用于后续操作附着
570 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
571 0 : ret = RunTemplate(tempAlg, level0ZeroCommInfo);
572 0 : CHK_PRT_RET(
573 : ret != HCCL_SUCCESS,
574 : HCCL_ERROR(
575 : "[CollAllGatherPipelineFor91093Executor][DoubleRingAllGather] Double ring "
576 : "AllGather failed, return[%d]",
577 : ret),
578 : ret);
579 : // 添加空task,保证执行时不乱序
580 0 : CHK_RET(AlgTemplateBase::ExecEmptyTask(inputMem, outputMem, stream, dispatcher_));
581 0 : return HCCL_SUCCESS;
582 0 : }
583 :
584 0 : HcclResult CollAllGatherPipelineFor91093Executor::GetSubStreamInfoOnOneRing(
585 : const u32 ringIndex, std::vector<Stream>& subStreamsInOneRing,
586 : std::vector<std::shared_ptr<LocalNotify>>& mainSignalsInOneRing,
587 : std::vector<std::shared_ptr<LocalNotify>>& subSignalsInOneRing)
588 : {
589 : // slaveStreams 不包含主流，先 +1 补回主流，再减去 2 条 Pipeline 专用流，得到基类 ring 数。
590 0 : const u32 ringNum = algResResp_->slaveStreams.size() + 1 - PIPELINE_EXTRA_STREAM_NUM;
591 0 : if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING) {
592 0 : subStreamsInOneRing.push_back(algResResp_->slaveStreams[ringIndex + 1]);
593 0 : mainSignalsInOneRing.push_back(algResResp_->notifiesMain[ringIndex + 1]);
594 0 : subSignalsInOneRing.push_back(algResResp_->notifiesAux[ringIndex + 1]);
595 0 : } else if (ringNum == LEVEL0_PLANE_NUM_IN_NPRING_SINGLE * STREAM_NUM_FOR_DMAREDUCE_ONE_RING) {
596 0 : subStreamsInOneRing.push_back(algResResp_->slaveStreams[ringIndex]);
597 0 : mainSignalsInOneRing.push_back(algResResp_->notifiesMain[ringIndex]);
598 0 : subSignalsInOneRing.push_back(algResResp_->notifiesAux[ringIndex]);
599 : }
600 0 : return HCCL_SUCCESS;
601 : }
602 :
603 : REGISTER_EXEC("AllGatherPipelineFor91093Executor", AllGatherPipelineFor91093, CollAllGatherPipelineFor91093Executor);
604 :
605 : } // namespace hccl
|
