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