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