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_executor.h"
12 : #include <numeric>
13 :
14 : namespace hccl {
15 12 : CollAllGatherExecutor::CollAllGatherExecutor(const HcclDispatcher dispatcher, std::unique_ptr<TopoMatcher>& topoMatcher)
16 12 : : CollCommExecutor(dispatcher, topoMatcher)
17 15 : {}
18 :
19 7 : HcclResult CollAllGatherExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
20 : {
21 7 : HcclUs startut = TIME_NOW();
22 7 : tag_ = param.tag;
23 7 : algResResp_ = &algRes;
24 :
25 7 : const u64 count = param.GetDataCount(topoAttr_.userRank);
26 7 : const HcclDataType dataType = param.GetDataType();
27 7 : bool needLaunchAtTheEnd = !is310P3Common_; // 是否需要在Orchestrate()结束时launch任务
28 :
29 7 : HcclResult ret = HCCL_SUCCESS;
30 : // 图模式和单卡场景下不需要Loop
31 7 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
32 5 : u64 countSize = count * SIZE_TABLE[dataType];
33 5 : u64 totalSize = CalcTotalCount(param) * SIZE_TABLE[dataType];
34 5 : ExecMem execMem;
35 5 : execMem.count = count;
36 5 : execMem.inputMem = DeviceMem::create(algRes.paramInputMem.ptr(), countSize);
37 5 : execMem.outputMem = DeviceMem::create(algRes.paramOutputMem.ptr(), totalSize);
38 5 : execMem.scratchMem = algRes.scratchMem;
39 5 : execMem.inputPtr = param.inputPtr;
40 5 : execMem.outputPtr = param.outputPtr;
41 5 : HCCL_DEBUG(
42 : "[CollAllGatherExecutor][Orchestrate]offload inputMem[%p][%llu], outputMem[%p][%llu],"
43 : "scratchMem[%p][%llu], inputPtr[%p] outputPtr[%p], count[%llu].",
44 : execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
45 : execMem.scratchMem.ptr(), execMem.scratchMem.size(), execMem.inputPtr, execMem.outputPtr, execMem.count);
46 5 : ret = KernelRun(param, execMem);
47 7 : } else if (topoAttr_.userRankSize == 1) {
48 0 : ExecMem execMem;
49 0 : execMem.count = count;
50 0 : execMem.inputMem = algRes.cclInputMem;
51 0 : execMem.outputMem = algRes.cclOutputMem;
52 0 : execMem.scratchMem = algRes.scratchMem;
53 0 : execMem.inputPtr = param.inputPtr;
54 0 : execMem.outputPtr = param.outputPtr;
55 0 : ret = KernelRun(param, execMem);
56 0 : needLaunchAtTheEnd = false;
57 2 : } else if (desc_.isZeroCopy) {
58 0 : u64 totalSize = param.DataDes.count * SIZE_TABLE[param.DataDes.dataType];
59 : // 在Level1和Level2执行RunLoop
60 0 : if (topoAttr_.serverNum > 1) {
61 0 : ret = RunLoop(param, algRes);
62 0 : CHK_PRT_RET(
63 : ret != HCCL_SUCCESS,
64 : HCCL_ERROR(
65 : "[CollAllGatherExecutor][Orchestrate]errNo[0x%016llx]AllGather executor run loop failed",
66 : HCCL_ERROR_CODE(ret)),
67 : ret);
68 : } else { // 单机场景，数据直接从UserInput搬到UserOutput
69 : DeviceMem dstMem = DeviceMem::create(
70 0 : static_cast<u8*>(algRes.paramOutputMem.ptr()) + totalSize * topoAttr_.userRank, totalSize);
71 0 : DeviceMem srcMem = DeviceMem::create(algRes.paramInputMem.ptr(), totalSize);
72 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
73 0 : }
74 : // 在Level0执行KernelRun
75 0 : ExecMem execMem;
76 0 : execMem.count = param.DataDes.count;
77 0 : execMem.inputMem = DeviceMem::create(algRes.paramInputMem.ptr(), totalSize);
78 0 : execMem.outputMem = DeviceMem::create(algRes.paramOutputMem.ptr(), totalSize * topoAttr_.userRankSize);
79 0 : execMem.scratchMem = algRes.scratchMem;
80 0 : execMem.inputPtr = param.inputPtr;
81 0 : execMem.outputPtr = param.outputPtr;
82 0 : ret = KernelRunIntraServerPost(param, execMem);
83 0 : } else {
84 2 : if (isAllGatherV_) {
85 0 : ret = RunLoopV(param, algRes);
86 : } else {
87 2 : ret = RunLoop(param, algRes);
88 : }
89 2 : needLaunchAtTheEnd = false;
90 : }
91 7 : CHK_PRT_RET(
92 : ret != HCCL_SUCCESS,
93 : HCCL_ERROR(
94 : "[CollAllGatherExecutor][Orchestrate]errNo[0x%016llx]AllGather executor kernel run failed",
95 : HCCL_ERROR_CODE(ret)),
96 : ret);
97 :
98 : // Enforce task launch at the end of Orchestrate
99 : // 注意: 不要删除这里的强制launch, 否则会导致aicpu cache功能问题
100 7 : if (needLaunchAtTheEnd) {
101 5 : HCCL_INFO("%s: enforce task launch at the end of Orchestrate", __func__);
102 5 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
103 : }
104 :
105 7 : HCCL_INFO(
106 : "tag[%s], Allgather executor orchestrate success, take time [%lld]us", param.tag.c_str(),
107 : DURATION_US(TIME_NOW() - startut));
108 7 : return HCCL_SUCCESS;
109 : }
110 :
111 0 : u64 CollAllGatherExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
112 : {
113 : // 中转内存单次最多能够接受的output count
114 0 : u64 maxCountPerLoop = cclBuffSize / topoAttr_.userRankSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
115 0 : HCCL_WARNING(
116 : "[CollAllGatherExecutor][CalcLoopMaxCount]"
117 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize",
118 : maxCountPerLoop);
119 0 : return maxCountPerLoop;
120 : }
121 :
122 18 : bool CollAllGatherExecutor::IsHugeData(const u64 curSize)
123 : {
124 36 : bool hugeData = curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE
125 18 : || curSize > SDMA_SEND_MAX_SIZE;
126 18 : return hugeData;
127 : }
128 :
129 2 : bool CollAllGatherExecutor::IsSmallData(const u64 size)
130 : {
131 2 : HCCL_INFO("[CollAllGatherExecutor][IsSmallData]opMeta is using the default option: not small data");
132 2 : return false;
133 : }
134 :
135 5 : u64 CollAllGatherExecutor::CalcTotalCount(const OpParam& param) const
136 : {
137 5 : if (isAllGatherV_) {
138 0 : const auto* countsPtr = static_cast<const u64*>(param.VDataDes.counts);
139 0 : return std::accumulate(countsPtr, countsPtr + topoAttr_.userRankSize, 0ULL);
140 : }
141 5 : return param.DataDes.count * topoAttr_.userRankSize;
142 : }
143 :
144 0 : bool CollAllGatherExecutor::CalcCountsDispls(
145 : const u64 maxTotalCount, std::vector<u64>& countsLeft, std::vector<u64>& displs, std::vector<u64>& curCounts,
146 : std::vector<u64>& curDispls)
147 : {
148 0 : bool finished = true;
149 :
150 0 : curCounts.resize(countsLeft.size(), 0);
151 0 : curDispls.resize(displs.size(), 0);
152 :
153 : // 先设置本轮的displacements，等于入参displs
154 0 : std::copy(displs.begin(), displs.end(), curDispls.begin());
155 :
156 : // 分配好每个rank的counts
157 0 : for (auto i = 0U; i < countsLeft.size(); ++i) {
158 0 : const auto curCount = countsLeft[i] < maxTotalCount ? countsLeft[i] : maxTotalCount;
159 0 : curCounts[i] = curCount;
160 0 : countsLeft[i] -= curCount;
161 0 : displs[i] += curCount;
162 :
163 0 : if (countsLeft[i] != 0) {
164 0 : finished = false;
165 : }
166 : }
167 :
168 0 : PrintCountsDispls(finished, curCounts, curDispls);
169 :
170 0 : return finished;
171 : }
172 :
173 0 : void CollAllGatherExecutor::PrintCountsDispls(
174 : bool finished, const std::vector<u64>& curCounts, const std::vector<u64>& curDispls)
175 : {
176 0 : if (HcclCheckLogLevel(DLOG_DEBUG)) {
177 0 : std::ostringstream curLoopInfo;
178 0 : curLoopInfo << "counts[ ";
179 0 : for (auto count : curCounts) {
180 0 : curLoopInfo << count << " ";
181 : }
182 0 : curLoopInfo << "], displs[ ";
183 0 : for (auto displ : curDispls) {
184 0 : curLoopInfo << displ << " ";
185 : }
186 0 : curLoopInfo << "]";
187 0 : HCCL_DEBUG(
188 : "[CollAllGatherExecutor][CountsDispls]finished[%u], Current loop info: %s", finished,
189 : curLoopInfo.str().c_str());
190 0 : }
191 0 : }
192 :
193 0 : std::vector<u64> CollAllGatherExecutor::GetCounts(const OpParam& param) const
194 : {
195 0 : const auto* countsPtr = static_cast<const u64*>(param.VDataDes.counts);
196 0 : return std::vector<u64>(countsPtr, countsPtr + topoAttr_.userRankSize);
197 : }
198 :
199 0 : std::vector<u64> CollAllGatherExecutor::GetDispls(const OpParam& param) const
200 : {
201 0 : const auto* displsPtr = static_cast<const u64*>(param.VDataDes.displs);
202 0 : return std::vector<u64>(displsPtr, displsPtr + topoAttr_.userRankSize);
203 : }
204 :
205 0 : u64 CollAllGatherExecutor::GetCurrentCount(const OpParam& param, const std::vector<u64>& curCounts) const
206 : {
207 0 : return curCounts[topoAttr_.userRank];
208 : }
209 :
210 0 : u64 CollAllGatherExecutor::CalcCurrentTotalCount(const OpParam& param, const std::vector<u64>& curCounts) const
211 : {
212 0 : return std::accumulate(curCounts.cbegin(), curCounts.cend(), 0ULL);
213 : }
214 :
215 : HcclOpMetaInfoDef
216 0 : CollAllGatherExecutor::GetOpMetaInfo(u32 algTypeLevel1, bool hugeData, bool smallData, bool dataSplit) const
217 : {
218 0 : return HcclOpMetaInfo::GetOneForAllGatherV(algTypeLevel1, hugeData, smallData, CopyPattern::BCOPY, dataSplit);
219 : }
220 :
221 0 : void CollAllGatherExecutor::UpdateOpParam(
222 : OpParam& param, std::vector<u64>& curCounts, std::vector<u64>& curDispls) const
223 : {
224 0 : param.VDataDes.counts = curCounts.data();
225 0 : param.VDataDes.displs = curDispls.data();
226 0 : }
227 :
228 : // 基于性能考量，合并RunLoop和RunLoopInner
229 2 : HcclResult CollAllGatherExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
230 : {
231 2 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
232 :
233 2 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
234 2 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
235 2 : void* commInputPtr = algRes.cclInputMem.ptr();
236 2 : u8* commOutputPtr = static_cast<u8*>(algRes.cclOutputMem.ptr());
237 2 : CHK_PTR_NULL(curInputPtr);
238 2 : CHK_PTR_NULL(curOutputPtr);
239 2 : CHK_PTR_NULL(commInputPtr);
240 2 : CHK_PTR_NULL(commOutputPtr);
241 :
242 2 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize);
243 2 : CHK_PRT_RET(
244 : maxCountPerLoop == 0,
245 : HCCL_ERROR(
246 : "[CollAllGatherExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
247 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
248 : HCCL_E_PARA);
249 :
250 2 : bool smallData = IsSmallData(param.DataDes.count * unitSize);
251 20 : for (u64 countLeft = param.DataDes.count, curCount = 0, inputOffset = 0, outputOffset = 0; countLeft > 0;
252 18 : countLeft -= curCount) {
253 18 : curInputPtr += inputOffset;
254 18 : curOutputPtr += outputOffset;
255 : // 判断剩余数据量对应的output size是否大于中转output size
256 18 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
257 18 : u64 curSize = curCount * unitSize; // 单位：字节
258 :
259 18 : HCCL_DEBUG(
260 : "[CollAllGatherExecutor][RunLoop]tag[%s], inputOffset[%llu], outputOffset[%llu], "
261 : "sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d]",
262 : param.tag.c_str(), inputOffset, outputOffset, curInputPtr, curOutputPtr, curCount, param.DataDes.dataType);
263 :
264 18 : if (!is310P3Common_) {
265 : /* 设置子图复用标志 */
266 18 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
267 18 : bool hugeData = IsHugeData(curSize); // override
268 18 : bool dataSplit = false;
269 18 : auto opMeta = HcclOpMetaInfo::GetOneForAllGather(
270 : autoSelectedAlgTypeLevel1, hugeData, smallData, CopyPattern::BCOPY, dataSplit);
271 18 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
272 : }
273 :
274 : // 执行
275 18 : if (!DMAReduceFlag_) {
276 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
277 0 : DeviceMem srcMem = DeviceMem::create(curInputPtr, curSize);
278 0 : DeviceMem dstMem = DeviceMem::create(commInputPtr, curSize);
279 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
280 0 : HCCL_DEBUG("[CollAllGatherExecutor][RunLoop]copy from user in to ccl in.");
281 0 : }
282 :
283 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
284 18 : ExecMem execMem;
285 18 : execMem.count = curCount;
286 18 : execMem.inputMem = DeviceMem::create(commInputPtr, curSize);
287 18 : u32 sliceNum = desc_.isZeroCopy ? topoAttr_.serverNum : topoAttr_.userRankSize;
288 18 : execMem.outputMem = DeviceMem::create(commOutputPtr, curSize * sliceNum);
289 18 : execMem.scratchMem = algRes.scratchMem;
290 18 : execMem.inputPtr = curInputPtr;
291 18 : execMem.outputPtr = curOutputPtr;
292 18 : HcclResult ret = HCCL_SUCCESS;
293 18 : if (!desc_.isZeroCopy) {
294 18 : ret = KernelRun(param, execMem);
295 : } else {
296 0 : ret = KernelRunInterServer(param, execMem);
297 : }
298 18 : CHK_PRT_RET(
299 : ret != HCCL_SUCCESS,
300 : HCCL_ERROR(
301 : "[CollAllGatherExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], "
302 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d].",
303 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr, curCount, param.DataDes.dataType),
304 : ret);
305 :
306 18 : if (!DMAReduceFlag_) {
307 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
308 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
309 : // 拷贝中转output上每个slice的数据到output内存，目的端中每个slice的size固定为output的size
310 0 : DeviceMem dstMem = DeviceMem::create(curOutputPtr + param.DataDes.count * unitSize * i, curSize);
311 0 : DeviceMem srcMem = DeviceMem::create(commOutputPtr + curSize * i, curSize);
312 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
313 0 : }
314 : }
315 :
316 18 : if (!is310P3Common_) {
317 18 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
318 : }
319 :
320 18 : inputOffset = curSize;
321 18 : outputOffset = curSize;
322 18 : }
323 2 : return HCCL_SUCCESS;
324 : }
325 :
326 0 : HcclResult CollAllGatherExecutor::RunLoopV(OpParam& param, AlgResourceResponse& algRes)
327 : {
328 0 : auto counts = GetCounts(param);
329 0 : auto displs = GetDispls(param);
330 0 : const HcclDataType dataType = param.GetDataType();
331 0 : u32 unitSize = SIZE_TABLE[dataType];
332 :
333 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
334 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
335 0 : u8* commInputPtr = static_cast<u8*>(algRes.cclInputMem.ptr());
336 0 : u8* commOutputPtr = static_cast<u8*>(algRes.cclOutputMem.ptr());
337 :
338 0 : if (UNLIKELY(curInputPtr == nullptr)) {
339 : // 若本rank的input count为0，此时允许curInputPtr传入空指针，为保证后续流程正常执行，赋值为cclin的地址
340 0 : curInputPtr = commInputPtr;
341 0 : HCCL_DEBUG("Since the input count is 0, set curInputPtr to ccl input[%p]", curInputPtr);
342 : } else {
343 0 : CHK_PTR_NULL(curInputPtr);
344 : }
345 0 : CHK_PTR_NULL(curOutputPtr);
346 0 : CHK_PTR_NULL(commInputPtr);
347 0 : CHK_PTR_NULL(commOutputPtr);
348 :
349 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize);
350 0 : CHK_PRT_RET(
351 : maxCountPerLoop == 0,
352 : HCCL_ERROR(
353 : "[CollAllGatherExecutor][RunLoopV]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
354 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
355 : HCCL_E_PARA);
356 :
357 0 : bool finished = false;
358 0 : while (!finished) {
359 0 : auto curCounts = std::vector<u64>();
360 0 : auto curDispls = std::vector<u64>();
361 : // 每轮loop需要重新计算counts和displs
362 0 : finished = CalcCountsDispls(maxCountPerLoop, counts, displs, curCounts, curDispls);
363 0 : u64 curCount = GetCurrentCount(param, curCounts);
364 0 : u64 curSize = curCount * unitSize; // 单位：字节
365 0 : const u64 totalSize = CalcCurrentTotalCount(param, curCounts) * unitSize;
366 :
367 0 : HCCL_DEBUG(
368 : "[CollAllGatherExecutor][RunLoopV]tag[%s], sendBuf[%p], recvBuf[%p], sendSize[%llu], "
369 : "recvSize[%llu], cclInputMem[%u], cclOutputMem[%u], dataType[%d]",
370 : param.tag.c_str(), curInputPtr, curOutputPtr, curSize, totalSize, algRes.cclInputMem.size(),
371 : algRes.cclOutputMem.size(), dataType);
372 :
373 : /* 设置子图复用标志 */
374 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
375 0 : bool hugeData = IsHugeData(curSize); // override
376 0 : bool smallData = IsSmallData(curSize);
377 0 : bool dataSplit = false;
378 0 : auto opMeta = GetOpMetaInfo(autoSelectedAlgTypeLevel1, hugeData, smallData, dataSplit);
379 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
380 :
381 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
382 0 : ExecMem execMem;
383 0 : execMem.count = curCount;
384 0 : execMem.inputMem = DeviceMem::create(commInputPtr, curSize);
385 0 : execMem.outputMem = DeviceMem::create(commOutputPtr, totalSize);
386 0 : execMem.scratchMem = algRes.scratchMem;
387 0 : execMem.inputPtr = curInputPtr;
388 0 : execMem.outputPtr = curOutputPtr;
389 0 : UpdateOpParam(param, curCounts, curDispls);
390 0 : HcclResult ret = KernelRun(param, execMem);
391 0 : CHK_PRT_RET(
392 : ret != HCCL_SUCCESS,
393 : HCCL_ERROR(
394 : "[CollAllGatherExecutor][RunLoopV]errNo[0x%016llx]kernel run error, tag[%s], "
395 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d]",
396 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr, curCount, dataType),
397 : ret);
398 :
399 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
400 :
401 0 : curInputPtr += curSize;
402 : // AllGatherV curOutputPtr不需要偏移，偏移由displs计算
403 0 : }
404 0 : return HCCL_SUCCESS;
405 0 : }
406 :
407 : HcclResult
408 18 : CollAllGatherExecutor::PrepareAllgatherSlice(u32 sliceNum, u64 inputMemSize, std::vector<Slice>& dataSegsSlice) const
409 : {
410 18 : Slice sliceTemp;
411 54 : for (u32 i = 0; i < sliceNum; i++) { // 根据数据量计算每个环上数据的偏移和大小
412 36 : sliceTemp.size = inputMemSize;
413 36 : sliceTemp.offset = inputMemSize * i;
414 36 : dataSegsSlice.push_back(sliceTemp);
415 : }
416 18 : return HCCL_SUCCESS;
417 : }
418 :
419 0 : HcclResult CollAllGatherExecutor::CalculateLevel1AllgatherSlice(
420 : u64 inputMemSize, u32 level0RankSize, u32 level1RankSize, std::vector<std::vector<Slice>> multRingsSliceZero,
421 : std::vector<std::vector<Slice>>& multRingsSlice) const
422 : {
423 0 : for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
424 0 : std::vector<Slice> level1DataSlice;
425 0 : for (u32 level0Idx = 0; level0Idx < level0RankSize; level0Idx++) {
426 0 : CHK_PRT_RET(
427 : multRingsSliceZero[ringIndex].size() < level0RankSize,
428 : HCCL_ERROR("[CalculateLevel1AllgatherSlice]multRingsSliceZero[ringIndex]"
429 : "size is smaller than level0RankSize."),
430 : HCCL_E_INTERNAL);
431 0 : for (u32 level1Idx = 0; level1Idx < level1RankSize; level1Idx++) {
432 0 : Slice tmpSlice;
433 0 : tmpSlice.size = multRingsSliceZero[ringIndex][level0Idx].size;
434 : tmpSlice.offset
435 0 : = multRingsSliceZero[ringIndex][level0Idx].offset + level1Idx * level0RankSize * inputMemSize;
436 0 : level1DataSlice.push_back(tmpSlice);
437 : }
438 : }
439 0 : multRingsSlice.push_back(level1DataSlice);
440 0 : }
441 0 : return HCCL_SUCCESS;
442 : }
443 :
444 36 : HcclResult CollAllGatherExecutor::CalculateLevel2AllgatherSlice(
445 : u64 inputMemSize, u32 level0RankSize, u32 level1RankSize, u32 level2RankSize,
446 : std::vector<std::vector<Slice>> multRingsSliceZero, std::vector<Slice>& level2DataSlice, u32 ringIndex) const
447 : {
448 108 : for (u32 level0Idx = 0; level0Idx < level0RankSize; level0Idx++) {
449 144 : for (u32 level2Idx = 0; level2Idx < level2RankSize; level2Idx++) {
450 216 : for (u32 level1Idx = 0; level1Idx < level1RankSize; level1Idx++) {
451 144 : Slice tmpSlice;
452 144 : tmpSlice.size = multRingsSliceZero[ringIndex][level0Idx].size;
453 : tmpSlice.offset
454 144 : = multRingsSliceZero[ringIndex][level0Idx].offset
455 144 : + (level1Idx * level0RankSize + level2Idx * level0RankSize * level1RankSize) * inputMemSize;
456 144 : level2DataSlice.push_back(tmpSlice);
457 : }
458 : }
459 : }
460 36 : return HCCL_SUCCESS;
461 : }
462 :
463 0 : HcclResult CollAllGatherExecutor::AllGatherLevel2(
464 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, u64 count, HcclDataType dataType, Stream& stream,
465 : HcomCollOpInfo* opInfo)
466 : {
467 0 : u32 perDataSize = 0;
468 0 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
469 :
470 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
471 0 : u32 commIndex = level0CommInfo.localRank;
472 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
473 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0));
474 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
475 :
476 0 : u64 inputMemSize = inputMem.size();
477 0 : u32 level0RankSize = level0CommInfo.localRankSize;
478 0 : u32 level1RankSize = level1CommInfo.localRankSize;
479 0 : u32 level2RankSize = level2CommInfo.localRankSize;
480 0 : u32 level0ServerIndex = level0CommInfo.localRank;
481 0 : u32 level1ServerIndex = level1CommInfo.localRank;
482 :
483 0 : std::unique_ptr<AlgTemplateBase> level2AGExecutor;
484 : level2AGExecutor
485 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
486 0 : HCCL_INFO("AllGather ring: using ring algo inter-server.");
487 0 : CHK_SMART_PTR_NULL(level2AGExecutor);
488 :
489 : // 计算slice, 不同超节点相同slice
490 0 : std::vector<Slice> level2DataSegsSlice;
491 0 : Slice sliceTemp;
492 0 : for (u32 i = 0; i < level2RankSize; i++) {
493 0 : sliceTemp.size = inputMemSize;
494 0 : sliceTemp.offset = i * level1RankSize * level0RankSize * inputMemSize;
495 0 : level2DataSegsSlice.push_back(sliceTemp);
496 : }
497 : // outputMem传整块，通过baseOffset偏移
498 0 : u64 level2BaseOffset = (level0ServerIndex + level1ServerIndex * level1RankSize) * inputMemSize;
499 0 : CHK_RET(level2AGExecutor->Prepare(
500 : outputMem, outputMem, inputMem, count, dataType, stream, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID,
501 : level2DataSegsSlice, level2BaseOffset));
502 :
503 0 : CHK_RET(level2AGExecutor->RegisterProfiler(
504 : (level2RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank, PROF_STAGE_0,
505 : HCCL_EXEC_STEP_NOT_SET, stream));
506 :
507 0 : CHK_RET(RunTemplate(level2AGExecutor, level2CommInfo));
508 0 : HCCL_INFO("AllGather double ring [superpod] level2 AllGather run success");
509 :
510 : // 第二步，各个AI Server 间 AllGather (ring/NHR)
511 0 : HCCL_INFO("commIdx:%u Tag[%s].commLevel1.size():%u", commIndex, tag.c_str(), level1RankSize);
512 :
513 0 : if (level1RankSize > 1) {
514 0 : std::unique_ptr<AlgTemplateBase> level1AGExecutor;
515 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
516 : level1AGExecutor
517 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
518 0 : HCCL_INFO("AllGather ring: using ring algo inter-server.");
519 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
520 : level1AGExecutor
521 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
522 0 : HCCL_INFO("AllGather ring: using nonuniform-bruck algo inter-server.");
523 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
524 : level1AGExecutor
525 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
526 0 : HCCL_INFO("AllGather ring: using nonuniform-hierarchical-ring algo inter-server.");
527 : } else {
528 0 : HCCL_ERROR("AllGather ring: unsupported algtype [%s].", AlgTypeToStr(algType_).c_str());
529 0 : return HCCL_E_NOT_SUPPORT;
530 : }
531 0 : CHK_SMART_PTR_NULL(level1AGExecutor);
532 :
533 : // 计算slice, 不同超节点相同slice
534 0 : std::vector<Slice> level1DataSegsSlice;
535 0 : for (u32 j = 0; j < level2RankSize; j++) {
536 0 : for (u32 i = 0; i < level1RankSize; i++) {
537 0 : sliceTemp.size = inputMemSize;
538 : sliceTemp.offset
539 0 : = (i * level0RankSize + j * level1RankSize * level0RankSize + level0ServerIndex) * inputMemSize;
540 0 : level1DataSegsSlice.push_back(sliceTemp);
541 : }
542 : }
543 :
544 0 : CHK_RET(level1AGExecutor->Prepare(
545 : outputMem, outputMem, inputMem, count, dataType, stream, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID,
546 : level1DataSegsSlice, 0));
547 :
548 0 : CHK_RET(level1AGExecutor->RegisterProfiler(
549 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank, PROF_STAGE_1,
550 : HCCL_EXEC_STEP_NOT_SET, stream));
551 :
552 0 : CHK_RET(RunTemplate(level1AGExecutor, level1CommInfo));
553 0 : HCCL_INFO("AllGather double ring [superpod] level1 AllGather run success");
554 0 : }
555 :
556 : // 节点内做AllGather double ring
557 0 : std::vector<Slice> dataSegsSlice;
558 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
559 0 : CHK_RET(PrepareAllgatherSlice(level0RankSize, inputMemSize, dataSegsSlice));
560 :
561 : // 多环数据切分
562 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
563 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, tag, false, topoAttr_.nicList);
564 : } else {
565 0 : multRingsSliceZero.push_back(dataSegsSlice);
566 : }
567 0 : std::vector<std::vector<Slice>> multRingsSlice;
568 0 : for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
569 0 : std::vector<Slice> level2DataSlice;
570 0 : CHK_RET(CalculateLevel2AllgatherSlice(
571 : inputMemSize, level0RankSize, level1RankSize, level2RankSize, multRingsSliceZero, level2DataSlice,
572 : ringIndex));
573 0 : multRingsSlice.push_back(level2DataSlice);
574 0 : }
575 :
576 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
577 0 : if (!DMAReduceFlag_) {
578 0 : multRingsUserMemSlice = multRingsSlice;
579 : } else {
580 0 : for (u32 ringIndex = 0; ringIndex < multRingsSlice.size(); ringIndex++) {
581 0 : std::vector<Slice> level2UserMemSlice;
582 0 : for (auto& cclSlice : multRingsSlice[ringIndex]) {
583 0 : Slice tmpSlice;
584 0 : tmpSlice.size = cclSlice.size;
585 : tmpSlice.offset
586 0 : = (cclSlice.offset / inputMemSize) * count * perDataSize + multRingsSliceZero[ringIndex][0].offset;
587 0 : level2UserMemSlice.push_back(tmpSlice);
588 0 : HCCL_DEBUG(
589 : "rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]", topoAttr_.userRank, ringIndex,
590 : tmpSlice.offset, tmpSlice.size);
591 : }
592 0 : multRingsUserMemSlice.push_back(level2UserMemSlice);
593 0 : }
594 : }
595 :
596 0 : CHK_RET(ActiveSlaveStreams(stream));
597 0 : if (DMAReduceFlag_ && level1RankSize > 1) {
598 : // AllGather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
599 0 : opInfo->inputAddr = nullptr;
600 : }
601 0 : CHK_RET(MultiRingAllGather(
602 : tag, inputMem, outputMem, count, dataType, multRingsSlice, stream, PROF_STAGE_2, 0, opInfo,
603 : multRingsUserMemSlice));
604 :
605 0 : HCCL_INFO("AllGather double ring [superpod] level2 AllGather run success");
606 0 : return HCCL_SUCCESS;
607 0 : }
608 :
609 : } // namespace hccl
|
