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([[maybe_unused]] 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(
206 : [[maybe_unused]] const OpParam& param, const std::vector<u64>& curCounts) const
207 : {
208 0 : return curCounts[topoAttr_.userRank];
209 : }
210 :
211 0 : u64 CollAllGatherExecutor::CalcCurrentTotalCount(
212 : [[maybe_unused]] const OpParam& param, const std::vector<u64>& curCounts) const
213 : {
214 0 : return std::accumulate(curCounts.cbegin(), curCounts.cend(), 0ULL);
215 : }
216 :
217 : HcclOpMetaInfoDef
218 0 : CollAllGatherExecutor::GetOpMetaInfo(u32 algTypeLevel1, bool hugeData, bool smallData, bool dataSplit) const
219 : {
220 0 : return HcclOpMetaInfo::GetOneForAllGatherV(algTypeLevel1, hugeData, smallData, CopyPattern::BCOPY, dataSplit);
221 : }
222 :
223 0 : void CollAllGatherExecutor::UpdateOpParam(
224 : OpParam& param, std::vector<u64>& curCounts, std::vector<u64>& curDispls) const
225 : {
226 0 : param.VDataDes.counts = curCounts.data();
227 0 : param.VDataDes.displs = curDispls.data();
228 0 : }
229 :
230 : // 基于性能考量，合并RunLoop和RunLoopInner
231 2 : HcclResult CollAllGatherExecutor::RunLoop(OpParam& param, AlgResourceResponse& algRes)
232 : {
233 2 : u32 unitSize = SIZE_TABLE[param.DataDes.dataType];
234 :
235 2 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
236 2 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
237 2 : void* commInputPtr = algRes.cclInputMem.ptr();
238 2 : u8* commOutputPtr = static_cast<u8*>(algRes.cclOutputMem.ptr());
239 2 : CHK_PTR_NULL(curInputPtr);
240 2 : CHK_PTR_NULL(curOutputPtr);
241 2 : CHK_PTR_NULL(commInputPtr);
242 2 : CHK_PTR_NULL(commOutputPtr);
243 :
244 2 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize);
245 2 : CHK_PRT_RET(
246 : maxCountPerLoop == 0,
247 : HCCL_ERROR(
248 : "[CollAllGatherExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
249 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
250 : HCCL_E_PARA);
251 :
252 2 : bool smallData = IsSmallData(param.DataDes.count * unitSize);
253 20 : for (u64 countLeft = param.DataDes.count, curCount = 0, inputOffset = 0, outputOffset = 0; countLeft > 0;
254 18 : countLeft -= curCount) {
255 18 : curInputPtr += inputOffset;
256 18 : curOutputPtr += outputOffset;
257 : // 判断剩余数据量对应的output size是否大于中转output size
258 18 : curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
259 18 : u64 curSize = curCount * unitSize; // 单位：字节
260 :
261 18 : HCCL_DEBUG(
262 : "[CollAllGatherExecutor][RunLoop]tag[%s], inputOffset[%llu], outputOffset[%llu], "
263 : "sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d]",
264 : param.tag.c_str(), inputOffset, outputOffset, curInputPtr, curOutputPtr, curCount, param.DataDes.dataType);
265 :
266 18 : if (!is310P3Common_) {
267 : /* 设置子图复用标志 */
268 18 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
269 18 : bool hugeData = IsHugeData(curSize); // override
270 18 : bool dataSplit = false;
271 18 : auto opMeta = HcclOpMetaInfo::GetOneForAllGather(
272 : autoSelectedAlgTypeLevel1, hugeData, smallData, CopyPattern::BCOPY, dataSplit);
273 18 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
274 : }
275 :
276 : // 执行
277 18 : if (!DMAReduceFlag_) {
278 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
279 0 : DeviceMem srcMem = DeviceMem::create(curInputPtr, curSize);
280 0 : DeviceMem dstMem = DeviceMem::create(commInputPtr, curSize);
281 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
282 0 : HCCL_DEBUG("[CollAllGatherExecutor][RunLoop]copy from user in to ccl in.");
283 0 : }
284 :
285 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
286 18 : ExecMem execMem;
287 18 : execMem.count = curCount;
288 18 : execMem.inputMem = DeviceMem::create(commInputPtr, curSize);
289 18 : u32 sliceNum = desc_.isZeroCopy ? topoAttr_.serverNum : topoAttr_.userRankSize;
290 18 : execMem.outputMem = DeviceMem::create(commOutputPtr, curSize * sliceNum);
291 18 : execMem.scratchMem = algRes.scratchMem;
292 18 : execMem.inputPtr = curInputPtr;
293 18 : execMem.outputPtr = curOutputPtr;
294 18 : HcclResult ret = HCCL_SUCCESS;
295 18 : if (!desc_.isZeroCopy) {
296 18 : ret = KernelRun(param, execMem);
297 : } else {
298 0 : ret = KernelRunInterServer(param, execMem);
299 : }
300 18 : CHK_PRT_RET(
301 : ret != HCCL_SUCCESS,
302 : HCCL_ERROR(
303 : "[CollAllGatherExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], "
304 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d].",
305 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr, curCount, param.DataDes.dataType),
306 : ret);
307 :
308 18 : if (!DMAReduceFlag_) {
309 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
310 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
311 : // 拷贝中转output上每个slice的数据到output内存，目的端中每个slice的size固定为output的size
312 0 : DeviceMem dstMem = DeviceMem::create(curOutputPtr + param.DataDes.count * unitSize * i, curSize);
313 0 : DeviceMem srcMem = DeviceMem::create(commOutputPtr + curSize * i, curSize);
314 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
315 0 : }
316 : }
317 :
318 18 : if (!is310P3Common_) {
319 18 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
320 : }
321 :
322 18 : inputOffset = curSize;
323 18 : outputOffset = curSize;
324 18 : }
325 2 : return HCCL_SUCCESS;
326 : }
327 :
328 0 : HcclResult CollAllGatherExecutor::RunLoopV(OpParam& param, AlgResourceResponse& algRes)
329 : {
330 0 : auto counts = GetCounts(param);
331 0 : auto displs = GetDispls(param);
332 0 : const HcclDataType dataType = param.GetDataType();
333 0 : u32 unitSize = SIZE_TABLE[dataType];
334 :
335 0 : u8* curInputPtr = static_cast<u8*>(param.inputPtr);
336 0 : u8* curOutputPtr = static_cast<u8*>(param.outputPtr);
337 0 : u8* commInputPtr = static_cast<u8*>(algRes.cclInputMem.ptr());
338 0 : u8* commOutputPtr = static_cast<u8*>(algRes.cclOutputMem.ptr());
339 :
340 0 : if (UNLIKELY(curInputPtr == nullptr)) {
341 : // 若本rank的input count为0，此时允许curInputPtr传入空指针，为保证后续流程正常执行，赋值为cclin的地址
342 0 : curInputPtr = commInputPtr;
343 0 : HCCL_DEBUG("Since the input count is 0, set curInputPtr to ccl input[%p]", curInputPtr);
344 : } else {
345 0 : CHK_PTR_NULL(curInputPtr);
346 : }
347 0 : CHK_PTR_NULL(curOutputPtr);
348 0 : CHK_PTR_NULL(commInputPtr);
349 0 : CHK_PTR_NULL(commOutputPtr);
350 :
351 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize);
352 0 : CHK_PRT_RET(
353 : maxCountPerLoop == 0,
354 : HCCL_ERROR(
355 : "[CollAllGatherExecutor][RunLoopV]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
356 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
357 : HCCL_E_PARA);
358 :
359 0 : bool finished = false;
360 0 : while (!finished) {
361 0 : auto curCounts = std::vector<u64>();
362 0 : auto curDispls = std::vector<u64>();
363 : // 每轮loop需要重新计算counts和displs
364 0 : finished = CalcCountsDispls(maxCountPerLoop, counts, displs, curCounts, curDispls);
365 0 : u64 curCount = GetCurrentCount(param, curCounts);
366 0 : u64 curSize = curCount * unitSize; // 单位：字节
367 0 : const u64 totalSize = CalcCurrentTotalCount(param, curCounts) * unitSize;
368 :
369 0 : HCCL_DEBUG(
370 : "[CollAllGatherExecutor][RunLoopV]tag[%s], sendBuf[%p], recvBuf[%p], sendSize[%llu], "
371 : "recvSize[%llu], cclInputMem[%u], cclOutputMem[%u], dataType[%d]",
372 : param.tag.c_str(), curInputPtr, curOutputPtr, curSize, totalSize, algRes.cclInputMem.size(),
373 : algRes.cclOutputMem.size(), dataType);
374 :
375 : /* 设置子图复用标志 */
376 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
377 0 : bool hugeData = IsHugeData(curSize); // override
378 0 : bool smallData = IsSmallData(curSize);
379 0 : bool dataSplit = false;
380 0 : auto opMeta = GetOpMetaInfo(autoSelectedAlgTypeLevel1, hugeData, smallData, dataSplit);
381 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
382 :
383 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
384 0 : ExecMem execMem;
385 0 : execMem.count = curCount;
386 0 : execMem.inputMem = DeviceMem::create(commInputPtr, curSize);
387 0 : execMem.outputMem = DeviceMem::create(commOutputPtr, totalSize);
388 0 : execMem.scratchMem = algRes.scratchMem;
389 0 : execMem.inputPtr = curInputPtr;
390 0 : execMem.outputPtr = curOutputPtr;
391 0 : UpdateOpParam(param, curCounts, curDispls);
392 0 : HcclResult ret = KernelRun(param, execMem);
393 0 : CHK_PRT_RET(
394 : ret != HCCL_SUCCESS,
395 : HCCL_ERROR(
396 : "[CollAllGatherExecutor][RunLoopV]errNo[0x%016llx]kernel run error, tag[%s], "
397 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d]",
398 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr, curCount, dataType),
399 : ret);
400 :
401 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
402 :
403 0 : curInputPtr += curSize;
404 : // AllGatherV curOutputPtr不需要偏移，偏移由displs计算
405 0 : }
406 0 : return HCCL_SUCCESS;
407 0 : }
408 :
409 : HcclResult
410 18 : CollAllGatherExecutor::PrepareAllgatherSlice(u32 sliceNum, u64 inputMemSize, std::vector<Slice>& dataSegsSlice) const
411 : {
412 18 : Slice sliceTemp;
413 54 : for (u32 i = 0; i < sliceNum; i++) { // 根据数据量计算每个环上数据的偏移和大小
414 36 : sliceTemp.size = inputMemSize;
415 36 : sliceTemp.offset = inputMemSize * i;
416 36 : dataSegsSlice.push_back(sliceTemp);
417 : }
418 18 : return HCCL_SUCCESS;
419 : }
420 :
421 0 : HcclResult CollAllGatherExecutor::CalculateLevel1AllgatherSlice(
422 : u64 inputMemSize, u32 level0RankSize, u32 level1RankSize, std::vector<std::vector<Slice>> multRingsSliceZero,
423 : std::vector<std::vector<Slice>>& multRingsSlice) const
424 : {
425 0 : for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
426 0 : std::vector<Slice> level1DataSlice;
427 0 : for (u32 level0Idx = 0; level0Idx < level0RankSize; level0Idx++) {
428 0 : CHK_PRT_RET(
429 : multRingsSliceZero[ringIndex].size() < level0RankSize,
430 : HCCL_ERROR("[CalculateLevel1AllgatherSlice]multRingsSliceZero[ringIndex]"
431 : "size is smaller than level0RankSize."),
432 : HCCL_E_INTERNAL);
433 0 : for (u32 level1Idx = 0; level1Idx < level1RankSize; level1Idx++) {
434 0 : Slice tmpSlice;
435 0 : tmpSlice.size = multRingsSliceZero[ringIndex][level0Idx].size;
436 : tmpSlice.offset
437 0 : = multRingsSliceZero[ringIndex][level0Idx].offset + level1Idx * level0RankSize * inputMemSize;
438 0 : level1DataSlice.push_back(tmpSlice);
439 : }
440 : }
441 0 : multRingsSlice.push_back(level1DataSlice);
442 0 : }
443 0 : return HCCL_SUCCESS;
444 : }
445 :
446 36 : HcclResult CollAllGatherExecutor::CalculateLevel2AllgatherSlice(
447 : u64 inputMemSize, u32 level0RankSize, u32 level1RankSize, u32 level2RankSize,
448 : std::vector<std::vector<Slice>> multRingsSliceZero, std::vector<Slice>& level2DataSlice, u32 ringIndex) const
449 : {
450 108 : for (u32 level0Idx = 0; level0Idx < level0RankSize; level0Idx++) {
451 144 : for (u32 level2Idx = 0; level2Idx < level2RankSize; level2Idx++) {
452 216 : for (u32 level1Idx = 0; level1Idx < level1RankSize; level1Idx++) {
453 144 : Slice tmpSlice;
454 144 : tmpSlice.size = multRingsSliceZero[ringIndex][level0Idx].size;
455 : tmpSlice.offset
456 144 : = multRingsSliceZero[ringIndex][level0Idx].offset
457 144 : + (level1Idx * level0RankSize + level2Idx * level0RankSize * level1RankSize) * inputMemSize;
458 144 : level2DataSlice.push_back(tmpSlice);
459 : }
460 : }
461 : }
462 36 : return HCCL_SUCCESS;
463 : }
464 :
465 0 : HcclResult CollAllGatherExecutor::AllGatherLevel2(
466 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, u64 count, HcclDataType dataType, Stream& stream,
467 : HcomCollOpInfo* opInfo)
468 : {
469 0 : u32 perDataSize = 0;
470 0 : CHK_RET(SalGetDataTypeSize(dataType, perDataSize));
471 :
472 0 : SubCommInfo level0CommInfo = GetSubCommInfo(COMM_LEVEL0, COMM_INDEX_0);
473 0 : u32 commIndex = level0CommInfo.localRank;
474 0 : SubCommInfo level1CommInfo = GetSubCommInfo(COMM_LEVEL1, commIndex);
475 0 : CHK_RET(CheckCommSize(COMM_LEVEL2, COMM_INDEX_0));
476 0 : SubCommInfo level2CommInfo = GetSubCommInfo(COMM_LEVEL2, COMM_INDEX_0);
477 :
478 0 : u64 inputMemSize = inputMem.size();
479 0 : u32 level0RankSize = level0CommInfo.localRankSize;
480 0 : u32 level1RankSize = level1CommInfo.localRankSize;
481 0 : u32 level2RankSize = level2CommInfo.localRankSize;
482 0 : u32 level0ServerIndex = level0CommInfo.localRank;
483 0 : u32 level1ServerIndex = level1CommInfo.localRank;
484 :
485 0 : std::unique_ptr<AlgTemplateBase> level2AGExecutor;
486 : level2AGExecutor
487 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
488 0 : HCCL_INFO("AllGather ring: using ring algo inter-server.");
489 0 : CHK_SMART_PTR_NULL(level2AGExecutor);
490 :
491 : // 计算slice, 不同超节点相同slice
492 0 : std::vector<Slice> level2DataSegsSlice;
493 0 : Slice sliceTemp;
494 0 : for (u32 i = 0; i < level2RankSize; i++) {
495 0 : sliceTemp.size = inputMemSize;
496 0 : sliceTemp.offset = i * level1RankSize * level0RankSize * inputMemSize;
497 0 : level2DataSegsSlice.push_back(sliceTemp);
498 : }
499 : // outputMem传整块，通过baseOffset偏移
500 0 : u64 level2BaseOffset = (level0ServerIndex + level1ServerIndex * level1RankSize) * inputMemSize;
501 0 : CHK_RET(level2AGExecutor->Prepare(
502 : outputMem, outputMem, inputMem, count, dataType, stream, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID,
503 : level2DataSegsSlice, level2BaseOffset));
504 :
505 0 : CHK_RET(level2AGExecutor->RegisterProfiler(
506 : (level2RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank, PROF_STAGE_0,
507 : HCCL_EXEC_STEP_NOT_SET, stream));
508 :
509 0 : CHK_RET(RunTemplate(level2AGExecutor, level2CommInfo));
510 0 : HCCL_INFO("AllGather double ring [superpod] level2 AllGather run success");
511 :
512 : // 第二步，各个AI Server 间 AllGather (ring/NHR)
513 0 : HCCL_INFO("commIdx:%u Tag[%s].commLevel1.size():%u", commIndex, tag.c_str(), level1RankSize);
514 :
515 0 : if (level1RankSize > 1) {
516 0 : std::unique_ptr<AlgTemplateBase> level1AGExecutor;
517 0 : if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_RING) {
518 : level1AGExecutor
519 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_RING, dispatcher_);
520 0 : HCCL_INFO("AllGather ring: using ring algo inter-server.");
521 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NB) {
522 : level1AGExecutor
523 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NB, dispatcher_);
524 0 : HCCL_INFO("AllGather ring: using nonuniform-bruck algo inter-server.");
525 0 : } else if (algType_.algoLevel1 == AlgTypeLevel1::ALG_LEVEL1_NHR) {
526 : level1AGExecutor
527 0 : = AlgTemplateRegistry::Instance().GetAlgTemplate(TemplateType::TEMPLATE_ALL_GATHER_NHR, dispatcher_);
528 0 : HCCL_INFO("AllGather ring: using nonuniform-hierarchical-ring algo inter-server.");
529 : } else {
530 0 : HCCL_ERROR("AllGather ring: unsupported algtype [%s].", AlgTypeToStr(algType_).c_str());
531 0 : return HCCL_E_NOT_SUPPORT;
532 : }
533 0 : CHK_SMART_PTR_NULL(level1AGExecutor);
534 :
535 : // 计算slice, 不同超节点相同slice
536 0 : std::vector<Slice> level1DataSegsSlice;
537 0 : for (u32 j = 0; j < level2RankSize; j++) {
538 0 : for (u32 i = 0; i < level1RankSize; i++) {
539 0 : sliceTemp.size = inputMemSize;
540 : sliceTemp.offset
541 0 : = (i * level0RankSize + j * level1RankSize * level0RankSize + level0ServerIndex) * inputMemSize;
542 0 : level1DataSegsSlice.push_back(sliceTemp);
543 : }
544 : }
545 :
546 0 : CHK_RET(level1AGExecutor->Prepare(
547 : outputMem, outputMem, inputMem, count, dataType, stream, HCCL_REDUCE_RESERVED, INVALID_VALUE_RANKID,
548 : level1DataSegsSlice, 0));
549 :
550 0 : CHK_RET(level1AGExecutor->RegisterProfiler(
551 : (level1RankSize << PROF_RANKSIZE_OFFSET_OF_PLANEID) + level2CommInfo.localRank, PROF_STAGE_1,
552 : HCCL_EXEC_STEP_NOT_SET, stream));
553 :
554 0 : CHK_RET(RunTemplate(level1AGExecutor, level1CommInfo));
555 0 : HCCL_INFO("AllGather double ring [superpod] level1 AllGather run success");
556 0 : }
557 :
558 : // 节点内做AllGather double ring
559 0 : std::vector<Slice> dataSegsSlice;
560 0 : std::vector<std::vector<Slice>> multRingsSliceZero; // 数据基于该rank上环0的偏移
561 0 : CHK_RET(PrepareAllgatherSlice(level0RankSize, inputMemSize, dataSegsSlice));
562 :
563 : // 多环数据切分
564 0 : if (topoType_ == TopoType::TOPO_TYPE_NP_DOUBLE_RING) {
565 0 : multRingsSliceZero = PrepareMultiRingSlice(dataSegsSlice, tag, false, topoAttr_.nicList);
566 : } else {
567 0 : multRingsSliceZero.push_back(dataSegsSlice);
568 : }
569 0 : std::vector<std::vector<Slice>> multRingsSlice;
570 0 : for (u32 ringIndex = 0; ringIndex < multRingsSliceZero.size(); ringIndex++) {
571 0 : std::vector<Slice> level2DataSlice;
572 0 : CHK_RET(CalculateLevel2AllgatherSlice(
573 : inputMemSize, level0RankSize, level1RankSize, level2RankSize, multRingsSliceZero, level2DataSlice,
574 : ringIndex));
575 0 : multRingsSlice.push_back(level2DataSlice);
576 0 : }
577 :
578 0 : std::vector<std::vector<Slice>> multRingsUserMemSlice;
579 0 : if (!DMAReduceFlag_) {
580 0 : multRingsUserMemSlice = multRingsSlice;
581 : } else {
582 0 : for (u32 ringIndex = 0; ringIndex < multRingsSlice.size(); ringIndex++) {
583 0 : std::vector<Slice> level2UserMemSlice;
584 0 : for (auto& cclSlice : multRingsSlice[ringIndex]) {
585 0 : Slice tmpSlice;
586 0 : tmpSlice.size = cclSlice.size;
587 : tmpSlice.offset
588 0 : = (cclSlice.offset / inputMemSize) * count * perDataSize + multRingsSliceZero[ringIndex][0].offset;
589 0 : level2UserMemSlice.push_back(tmpSlice);
590 0 : HCCL_DEBUG(
591 : "rank[%u], ringIndex[%u], tmpSlice.offset=[%llu], size=[%llu]", topoAttr_.userRank, ringIndex,
592 : tmpSlice.offset, tmpSlice.size);
593 : }
594 0 : multRingsUserMemSlice.push_back(level2UserMemSlice);
595 0 : }
596 : }
597 :
598 0 : CHK_RET(ActiveSlaveStreams(stream));
599 0 : if (DMAReduceFlag_ && level1RankSize > 1) {
600 : // AllGather输入放在CCL buffer上，通过设置nullptr指示要从CCL buffer获取输入
601 0 : opInfo->inputAddr = nullptr;
602 : }
603 0 : CHK_RET(MultiRingAllGather(
604 : tag, inputMem, outputMem, count, dataType, multRingsSlice, stream, PROF_STAGE_2, 0, opInfo,
605 : multRingsUserMemSlice));
606 :
607 0 : HCCL_INFO("AllGather double ring [superpod] level2 AllGather run success");
608 0 : return HCCL_SUCCESS;
609 0 : }
610 :
611 : } // namespace hccl
|
