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_v_executor.h"
12 :
13 : namespace hccl {
14 0 : CollAllGatherVExecutor::CollAllGatherVExecutor(const HcclDispatcher dispatcher,
15 0 : std::unique_ptr<TopoMatcher> &topoMatcher)
16 0 : : CollCommExecutor(dispatcher, topoMatcher)
17 : {
18 0 : }
19 :
20 0 : HcclResult CollAllGatherVExecutor::Orchestrate(OpParam& param, AlgResourceResponse& algRes)
21 : {
22 0 : HcclUs startut = TIME_NOW();
23 0 : tag_ = param.tag;
24 0 : algResResp_ = &algRes;
25 :
26 0 : u64 count = static_cast<u64*>(param.VDataDes.counts)[topoAttr_.userRank];
27 0 : HcclResult ret = HCCL_SUCCESS;
28 : // 图模式场景下不需要Loop
29 0 : if (workflowMode_ != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
30 0 : ExecMem execMem;
31 0 : execMem.count = count;
32 0 : execMem.inputMem = algRes.paramInputMem;
33 0 : execMem.outputMem = algRes.paramOutputMem;
34 0 : execMem.scratchMem = algRes.scratchMem;
35 0 : execMem.inputPtr = param.inputPtr;
36 0 : execMem.outputPtr = param.outputPtr;
37 0 : HCCL_DEBUG("[CollAllGatherVExecutor][Orchestrate]offload inputMem[%p][%llu], outputMem[%p][%llu]," \
38 : "scratchMem[%p][%llu], inputPtr[%p] outputPtr[%p], count[%llu]",
39 : execMem.inputMem.ptr(), execMem.inputMem.size(), execMem.outputMem.ptr(), execMem.outputMem.size(),
40 : execMem.scratchMem.ptr(), execMem.scratchMem.size(), execMem.inputPtr, execMem.outputPtr, execMem.count);
41 0 : ret = KernelRun(param, execMem);
42 0 : } else {
43 0 : ret = RunLoop(param, algRes);
44 : }
45 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
46 : HCCL_ERROR("[CollAllGatherVExecutor][Orchestrate]errNo[0x%016llx]AllGatherV executor kernel run failed",
47 : HCCL_ERROR_CODE(ret)), ret);
48 0 : HCCL_INFO("tag[%s], AllgatherV executor orchestrate success, take time [%lld]us.",
49 : param.tag.c_str(), DURATION_US(TIME_NOW() - startut));
50 0 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult CollAllGatherVExecutor::GetAdjInfo(AlgResourceResponse& algRes, AdjInfo& adjInfo)
54 : {
55 : (void) algRes;
56 : (void) adjInfo;
57 0 : return HCCL_SUCCESS;
58 : }
59 :
60 0 : u64 CollAllGatherVExecutor::CalcLoopMaxCount(const u64 cclBuffSize, const u32 unitSize)
61 : {
62 : // 中转内存单次最多能够接受的output count
63 0 : u64 maxCountPerLoop = cclBuffSize / HCCL_MIN_SLICE_ALIGN * HCCL_MIN_SLICE_ALIGN / unitSize;
64 :
65 0 : HCCL_WARNING("[CollAllGatherVExecutor][CalcLoopMaxCount]" \
66 : "using default maxCountPerLoop[%llu] as CCLBuffSize / unitSize.", maxCountPerLoop);
67 0 : return maxCountPerLoop;
68 : }
69 :
70 0 : bool CollAllGatherVExecutor::IsHugeData(const u64 curSize)
71 : {
72 0 : bool hugeData = curSize * topoAttr_.userRankSize / HCCL_INTERNODE_MAX_DATA_RATE > RDMA_SEND_MAX_SIZE ||
73 : curSize > SDMA_SEND_MAX_SIZE;
74 0 : return hugeData;
75 : }
76 :
77 0 : HcclResult CollAllGatherVExecutor::CalcCurCountsAndCurDispls(const u64 maxTotalCount, std::vector<u64> &countsLeft,
78 : std::vector<u64> &displs, std::vector<u64> &curCounts, std::vector<u64> &curDispls, bool &finished)
79 : {
80 0 : HCCL_DEBUG("[CollAllGatherVExecutor][CalcCurCountsAndCurDispls]default func called.");
81 0 : return HCCL_SUCCESS;
82 : }
83 :
84 0 : HcclResult CollAllGatherVExecutor::RunLoop(OpParam ¶m, AlgResourceResponse &algRes)
85 : {
86 : // 每轮loop需要重新计算counts和displs
87 0 : const auto *countsPtr = static_cast<const u64*>(param.VDataDes.counts);
88 0 : auto countsLeft = std::vector<u64>(countsPtr, countsPtr + topoAttr_.userRankSize);
89 0 : const auto *displsPtr = static_cast<const u64*>(param.VDataDes.displs);
90 0 : auto displs = std::vector<u64>(displsPtr, displsPtr + topoAttr_.userRankSize);
91 :
92 0 : const HcclDataType dataType = param.VDataDes.dataType;
93 0 : u32 unitSize = SIZE_TABLE[dataType];
94 :
95 0 : u8 *curInputPtr = static_cast<u8 *>(param.inputPtr);
96 0 : u8 *curOutputPtr = static_cast<u8 *>(param.outputPtr);
97 0 : u8 *commInputPtr = static_cast<u8 *>(algRes.cclInputMem.ptr());
98 0 : u8 *commOutputPtr = static_cast<u8 *>(algRes.cclOutputMem.ptr());
99 :
100 0 : if (UNLIKELY(countsLeft[topoAttr_.userRank] == 0 && curInputPtr == nullptr)) {
101 : // 若本rank的input count为0，此时允许curInputPtr传入空指针，为保证后续流程正常执行，赋值为cclin的地址
102 0 : curInputPtr = commInputPtr;
103 0 : HCCL_DEBUG("Since the input count is 0, set curInputPtr to ccl input[%p]", curInputPtr);
104 : } else {
105 0 : CHK_PTR_NULL(curInputPtr);
106 : }
107 :
108 0 : CHK_PTR_NULL(curOutputPtr);
109 0 : CHK_PTR_NULL(commInputPtr);
110 0 : CHK_PTR_NULL(commOutputPtr);
111 :
112 : // 计算MaxCountPerLoop
113 0 : u64 maxCountPerLoop = CalcLoopMaxCount(algRes.cclInputMem.size(), unitSize); // override
114 0 : CHK_PRT_RET(maxCountPerLoop == 0,
115 : HCCL_ERROR("[CollAllGatherVExecutor][RunLoop]tag[%s], userRankSize is [%u], maxCountPerLoop is [%llu].",
116 : param.tag.c_str(), topoAttr_.userRankSize, maxCountPerLoop),
117 : HCCL_E_PARA);
118 :
119 0 : bool finished = false;
120 0 : while (!finished) {
121 : // 每个块尽可能平分，以均衡利用带宽
122 0 : auto curCounts = std::vector<u64>();
123 0 : auto curDispls = std::vector<u64>();
124 0 : CHK_RET(CalcCurCountsAndCurDispls(maxCountPerLoop, countsLeft, displs, curCounts, curDispls, finished));
125 :
126 : // 打印调测信息
127 0 : PrintCurCountAndCurDispls(curCounts, curDispls);
128 0 : u64 totalCount = 0;
129 0 : CHK_RET(CalcTotalCount(curCounts, totalCount));
130 0 : u64 OutputSize = totalCount * unitSize; // 单位：字节
131 :
132 0 : u64 curCount = curCounts[topoAttr_.userRank];
133 0 : u64 InputSize = curCount * unitSize;
134 0 : u64 curMaxCount = *std::max_element(curCounts.begin(), curCounts.end());
135 0 : u64 curMaxSize = curMaxCount * unitSize;
136 0 : HCCL_DEBUG("[CollAllGatherVExecutor][RunLoop]tag[%s], sendBuf[%p], recvBuf[%p], sendCount[%llu], dataType[%d], "
137 : "OutputSize[%llu], curMaxSize[%llu].",
138 : param.tag.c_str(), curInputPtr, curOutputPtr, curCount, dataType, OutputSize, curMaxSize);
139 :
140 0 : if (!is310P3Common_) {
141 : /* 设置子图复用标志 */
142 0 : auto autoSelectedAlgTypeLevel1 = static_cast<u32>(algType_.algoLevel1);
143 0 : bool hugeData = IsHugeData(curMaxSize); // override
144 0 : auto opMeta = HcclOpMetaInfo::GetOneForAllGatherV(autoSelectedAlgTypeLevel1, hugeData, false,
145 : CopyPattern::BCOPY, false);
146 0 : CHK_RET(InitTask(dispatcher_, param.stream, opMeta.isEnableCache, opMeta.GetCacheKey()));
147 : }
148 :
149 : // 执行
150 0 : if (!DMAReduceFlag_) {
151 : // 如果使用in CCL buffer，需要将user buffer in中的结果拷贝到CCL buffer in
152 0 : DeviceMem srcMem = DeviceMem::create(curInputPtr, InputSize);
153 0 : DeviceMem dstMem = DeviceMem::create(commInputPtr, InputSize);
154 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
155 0 : HCCL_DEBUG("[CollAllGatherVExecutor][RunLoop]copy from user in to ccl in.");
156 0 : }
157 :
158 : // 使用当前Loop偏移到的地址作为当前的inputPtr和outputPtr
159 0 : ExecMem execMem;
160 0 : execMem.count = curCount;
161 0 : execMem.inputMem = algRes.cclInputMem;
162 0 : execMem.outputMem = algRes.cclOutputMem;
163 0 : execMem.scratchMem = algRes.scratchMem;
164 0 : execMem.inputPtr = curInputPtr;
165 0 : execMem.outputPtr = curOutputPtr;
166 :
167 0 : OpParam curParam = param;
168 0 : curParam.VDataDes.counts = curCounts.data();
169 0 : curParam.VDataDes.displs = curDispls.data();
170 0 : curParam.VDataDes.dataType = dataType;
171 0 : HcclResult ret = KernelRun(curParam, execMem);
172 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
173 : HCCL_ERROR("[CollAllGatherVExecutor][RunLoop]errNo[0x%016llx]kernel run error, tag[%s], " \
174 : "inputMem ptr[%p], outputMem ptr[%p], count[%llu], dataType[%d]",
175 : HCCL_ERROR_CODE(ret), param.tag.c_str(), commInputPtr, commOutputPtr,
176 : curCount, dataType),
177 : ret);
178 :
179 0 : if (!DMAReduceFlag_) {
180 0 : u64 offSetCount = 0;
181 : // 如果使用CCL buffer，需要将CCL buffer out中的结果拷贝到user buffer out
182 0 : for (u32 i = 0; i < topoAttr_.userRankSize; i++) {
183 : // 拷贝中转output上每个slice的数据到output内存，目的端中每个slice的size固定为output的size
184 0 : DeviceMem dstMem = DeviceMem::create(curOutputPtr + curDispls[i] * unitSize, curCounts[i] * unitSize);
185 0 : DeviceMem srcMem = DeviceMem::create(commOutputPtr + offSetCount * unitSize, curCounts[i] * unitSize);
186 0 : offSetCount += curCounts[i];
187 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, dstMem, srcMem, param.stream));
188 0 : }
189 : }
190 :
191 0 : if (!is310P3Common_) {
192 0 : CHK_RET(LaunchTaskExtend(dispatcher_, param.stream, algResResp_->slaveStreams));
193 : }
194 :
195 0 : curInputPtr += InputSize;
196 : // AllGatherV curOutputPtr不需要偏移，output的偏移由displs计算
197 0 : }
198 0 : return HCCL_SUCCESS;
199 0 : }
200 :
201 0 : void CollAllGatherVExecutor::PrintCurCountAndCurDispls(const std::vector<u64> &curCounts,
202 : const std::vector<u64> &curDispls)
203 : {
204 0 : if (HcclCheckLogLevel(DLOG_DEBUG)) {
205 0 : std::ostringstream curLoopInfo;
206 0 : curLoopInfo << "Counts[ ";
207 0 : for (auto count : curCounts) {
208 0 : curLoopInfo << count << " ";
209 : }
210 0 : curLoopInfo << "], displs[ ";
211 0 : for (auto displ : curDispls) {
212 0 : curLoopInfo << displ << " ";
213 : }
214 0 : curLoopInfo << "]";
215 0 : HCCL_DEBUG("[CollAllGatherVExecutor][PrintCurCountAndCurDispls] Current loop info: %s",
216 : curLoopInfo.str().c_str());
217 0 : }
218 0 : }
219 :
220 0 : HcclResult CollAllGatherVExecutor::CalcTotalCount(std::vector<u64> curCounts, u64 &totalCount)
221 : {
222 0 : for(u64 i = 0; i < topoAttr_.userRankSize; i++)
223 : {
224 0 : totalCount += curCounts[i];
225 : }
226 :
227 0 : return HCCL_SUCCESS;
228 : }
229 :
230 : } // namespace hccl
|
