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 : #include <algorithm>
11 : #include <unordered_set>
12 : #include "mc2_compont.h"
13 : #include "ccu_assist.h"
14 : #include "mc2_context.h"
15 : #include "ccu_task_arg_mc2.h"
16 : #include "coll_service_device_mode.h"
17 : #include "mc2_global_mirror_tasks.h"
18 : #include "op_params_checker.h"
19 :
20 : namespace Hccl {
21 :
22 256 : Mc2Compont::~Mc2Compont()
23 : {
24 256 : auto deviceLogicId = comm->GetDeviceLogicId();
25 263 : for (const auto &server : ccuServerMap) {
26 7 : auto ret = InsExeQue::DeregisterExtendInstruction(deviceLogicId, server.first);
27 21 : HCCL_INFO("[Mc2Compont:%s]Destroy ccuServer execId[%u]", __func__, server.first);
28 7 : if (ret != HcclResult::HCCL_SUCCESS) {
29 15 : HCCL_ERROR("DeregisterExtendInstruction execId[%u] failed, ret[%d]", server.first, ret);
30 : }
31 : }
32 256 : ccuServerMap.clear();
33 256 : }
34 :
35 2 : void Mc2Compont::AllocCommResource(void *mc2Tiling, void **commContext)
36 : {
37 2 : auto tilingVersion = *static_cast<uint32_t *>(mc2Tiling);
38 6 : HCCL_INFO("[Mc2Compont:%s] Tiling version [%u]", __func__, tilingVersion);
39 2 : if (tilingVersion != UNKNOWN_TILING_V1 && tilingVersion != UNKNOWN_TILING_V2) {
40 0 : THROW<NotSupportException>(StringFormat("Tiling version not support, version[%u]", tilingVersion));
41 : }
42 :
43 2 : if (comm->GetRankSize() == 1) {
44 3 : HCCL_WARNING("Comm[%s] rank size is 1, Mc2 not support", comm->GetId().c_str());
45 1 : return;
46 : }
47 :
48 1 : std::unordered_set<uint64_t> algoTemplateRequire;
49 1 : if (tilingVersion == UNKNOWN_TILING_V1) {
50 : // 申请deviceMem、通信域信息获取、commContext赋值
51 0 : Alloc();
52 : // 生成本次需要的算子模板
53 0 : GenerateAlgoTemplates(reinterpret_cast<Mc2Tiling *>(mc2Tiling), algoTemplateRequire);
54 : } else {
55 : // 申请deviceMem、通信域信息获取、commContext赋值
56 1 : AllocV2();
57 : // 生成本次需要的算子模板
58 1 : GenerateAlgoTemplatesV2(reinterpret_cast<Mc2InitTilingInner *>(mc2Tiling), algoTemplateRequire);
59 : }
60 :
61 3 : HCCL_RUN_INFO("hcclCombinOpParam info: workSpace = [%llu], rankId = [%u], rankDim = [%u], xnAddr = [%llu], "
62 : "ckeAddr = [%llu], winSize = [%llu], windowsOut[0] = [%llu]",
63 : combinOpParam.workSpace, combinOpParam.rankId, combinOpParam.rankDim, combinOpParam.xnAddr,
64 : combinOpParam.ckeAddr, combinOpParam.winSize, combinOpParam.windowsOut[0]);
65 3 : HCCL_RUN_INFO("opType[0] = [%u], opType[1] = [%u], opType[2] = [%u], opType[3] = [%u], opType[4] = [%u], "
66 : "opType[5] = [%u], opType[6] = [%u], opType[7] = [%u], ", combinOpParam.opType[0],
67 : combinOpParam.opType[1], combinOpParam.opType[2], combinOpParam.opType[3], combinOpParam.opType[4],
68 : combinOpParam.opType[5], combinOpParam.opType[6], combinOpParam.opType[7]);
69 3 : HCCL_RUN_INFO("algorithmType[0] = [%u], algorithmType[1] = [%u], algorithmType[2] = [%u], algorithmType[3] = [%u], "
70 : "algorithmType[4] = [%u], algorithmType[5] = [%u], algorithmType[6] = [%u], algorithmType[7] = [%u]",
71 : combinOpParam.algorithmType[0], combinOpParam.algorithmType[1], combinOpParam.algorithmType[2],
72 : combinOpParam.algorithmType[3], combinOpParam.algorithmType[4], combinOpParam.algorithmType[5],
73 : combinOpParam.algorithmType[6], combinOpParam.algorithmType[7]);
74 1 : auto paramSize = sizeof(HcclCombinOpParam);
75 1 : if(combinOpParamBuffer == nullptr) {
76 1 : combinOpParamBuffer = std::make_shared<DevBuffer>(paramSize);
77 : }
78 1 : HrtMemcpy(reinterpret_cast<void *>(combinOpParamBuffer->GetAddr()), paramSize, static_cast<void *>(&combinOpParam),
79 : paramSize, RT_MEMCPY_HOST_TO_DEVICE);
80 1 : *commContext = reinterpret_cast<void *>(combinOpParamBuffer->GetAddr());
81 : // 生成ccuServer指令，将注册得到的execId保存在curExecId，GetCcuTaskInfo时通过curExecId获取TaskParam
82 1 : GenerateCcuServer(algoTemplateRequire);
83 1 : }
84 :
85 2 : static bool GetArgSizeFlag(std::vector<std::vector<CcuTaskParam>> &taskParams)
86 : {
87 : /*
88 : * 在MC2场景下，暂定只支持三种场景：1）单die且一个mission，2）双die且每die一个mission，3）单die且多个mission
89 : * 一次调用HCCL接口只能支持一种场景，其中2）包括所有双die算法，3）包括带尾块处理的算法与HalfAlltoAllV算子，1）包括其余算法
90 : * 只有场景3）时需要使argSize=1，1）和2）argSize均不变
91 : */
92 2 : std::unordered_set<uint8_t> dieIdSet;
93 2 : std::unordered_set<uint8_t> missionIdSet;
94 4 : for (auto &task : taskParams) {
95 2 : dieIdSet.emplace(task[0].dieId);
96 2 : missionIdSet.emplace(task[0].missionId);
97 6 : HCCL_INFO("TaskParam: dieId = [%u], missionId = [%u]", task[0].dieId, task[0].missionId);
98 : }
99 2 : bool oneDieOneMission = (dieIdSet.size() == 1) && (missionIdSet.size() == 1);
100 2 : bool twoDieOneMission = (dieIdSet.size() == 2) && (missionIdSet.size() == 1);
101 2 : bool oneDieMultiMission = (dieIdSet.size() == 1) && (missionIdSet.size() > 1);
102 2 : if (!oneDieOneMission && !twoDieOneMission && !oneDieMultiMission) {
103 0 : THROW<NotSupportException>(
104 0 : StringFormat("MC2 Scene cannot support: not OneDieOneMission, TwoDieOneMission or OneDieMultiMission !"));
105 : }
106 2 : return oneDieMultiMission;
107 2 : }
108 :
109 2 : std::vector<CcuTaskParam> Mc2Compont::GetCcuTaskInfo(void *tilingData)
110 : {
111 6 : HCCL_INFO("%s start.", __func__);
112 6 : HCCL_INFO("tilingData=%llu", tilingData);
113 2 : std::vector<std::vector<CcuTaskParam>> taskParams;
114 2 : std::vector<CcuTaskParam> ccuTaskParam;
115 2 : if(tilingData == nullptr) {
116 0 : return ccuTaskParam;
117 : }
118 2 : auto mc2Tiling = reinterpret_cast<Mc2Tiling *>(tilingData);
119 6 : HCCL_INFO("mc2Tiling=%s", mc2Tiling->ToString().c_str());
120 2 : auto version = mc2Tiling->version;
121 :
122 2 : if (version != UNKNOWN_TILING_V1 && version != UNKNOWN_TILING_V2) {
123 0 : THROW<NotSupportException>(StringFormat("Tiling version not support, version[%u]", version));
124 : }
125 :
126 : // 校验curExecId是否有效
127 2 : if (ccuServerMap.find(curExecId) == ccuServerMap.end()) {
128 0 : THROW<Hccl::InternalException>(
129 0 : StringFormat("CcuServer not find, curExecId[%llu], ccuServerSize[%d]", curExecId, ccuServerMap.size()));
130 : }
131 :
132 2 : CcuTaskArgMc2 ccuTaskArg(tokenInfo);
133 2 : HcclResult ret = CcuCtxMgr::GetTaskParam(comm->GetDeviceLogicId(), ccuTaskArg, curExecId, taskParams);
134 2 : if (ret != HcclResult::HCCL_SUCCESS) {
135 0 : THROW<CcuApiException>(StringFormat("GetTaskParam failed, ret[%u]", ret));
136 : }
137 2 : bool argSizeFlag = GetArgSizeFlag(taskParams);
138 3 : for (auto &task : taskParams) {
139 2 : if (task.size() != 1) {
140 1 : THROW<Hccl::InternalException>(
141 3 : StringFormat("Task Num In TaskParams Should Be 1, While It Is %d", task.size()));
142 : }
143 1 : if (argSizeFlag) {
144 0 : task[0].argSize = 1;
145 : }
146 1 : u32 notifyTimeout = comm->GetNotifyTimeoutCfg().GetNotifyTimeout();
147 1 : task[0].timeout = (notifyTimeout > UINT16_MAX) ? static_cast<uint16_t>(UINT16_MAX) : static_cast<uint16_t>(notifyTimeout);
148 1 : ccuTaskParam.push_back(task[0]);
149 1 : SaveMc2DfxTaskInfo(task[0], curExecId);
150 : }
151 1 : std::sort(ccuTaskParam.begin(), ccuTaskParam.end(), [](const CcuTaskParam &a, const CcuTaskParam &b) {
152 0 : return a.missionId < b.missionId;
153 : });
154 3 : HCCL_INFO("GetCcuTaskInfo success");
155 1 : return ccuTaskParam;
156 4 : }
157 :
158 1 : void Mc2Compont::Alloc()
159 : {
160 : // inputMem给算法编排使用，只需要申请一次，按照最大数据类型申请
161 1 : inputMem = std::make_shared<DevBuffer>(dataCount * DataTypeSizeGet(DataType::INT64) * comm->GetRankSize());
162 3 : HCCL_INFO("[Mc2Compont][Alloc]inputMem addr[%p] size = [%llu]", inputMem->GetAddr(), inputMem->GetSize());
163 9 : for(uint32_t i = 0; i < MAX_OP_NUM; i++) {
164 8 : combinOpParam.opType[i] = 0;
165 8 : combinOpParam.algorithmType[i] = 0;
166 : }
167 1 : if (ccuResourceAlloced) {
168 0 : return;
169 : }
170 :
171 1 : constexpr uint32_t comSyncNum = 2; // 每轮同步使用2个同步信号
172 1 : uint32_t comParamBufSize = CCU_TASK_NUM_MAX * CCU_PARAM_NUM_MAX * CCU_ONE_PARAM_SIZE ;
173 1 : uint32_t comSyncBufSize = CCU_TASK_NUM_MAX * comSyncNum * CCU_ONE_PARAM_SIZE ;
174 1 : workspaceBuffer = std::make_shared<DevBuffer>(MC2_WORKSPACE_SIZE);
175 1 : comParamBuffer = std::make_shared<DevBuffer>(comParamBufSize);
176 1 : comSyncBuffer = std::make_shared<DevBuffer>(comSyncBufSize);
177 :
178 1 : combinOpParam.workSpace = static_cast<uint64_t>(workspaceBuffer->GetAddr());
179 1 : combinOpParam.workSpaceSize = MC2_WORKSPACE_SIZE;
180 1 : combinOpParam.rankId = comm->GetMyRank();
181 1 : combinOpParam.rankDim = comm->GetRankSize();
182 1 : combinOpParam.xnAddr = static_cast<uint64_t>(comParamBuffer->GetAddr());
183 1 : combinOpParam.ckeAddr = static_cast<uint64_t>(comSyncBuffer->GetAddr());
184 : // add cclbuffer info
185 1 : if (comm->GetCclBuffer() == nullptr) {
186 0 : THROW<Hccl::InternalException>(StringFormat("Cannot get CCL Buffer to fill window!"));
187 : }
188 1 : combinOpParam.winSize = static_cast<uint64_t>(comm->GetCclBuffer()->GetSize());
189 1 : combinOpParam.windowsOut[0] = static_cast<uint64_t>(comm->GetCclBuffer()->GetAddr());
190 1 : ccuResourceAlloced = true;
191 :
192 1 : tokenInfo = CcuRep::GetTokenInfo(static_cast<uint64_t>(workspaceBuffer->GetAddr()),
193 1 : static_cast<uint64_t>(workspaceBuffer->GetSize()));
194 : }
195 :
196 1 : void Mc2Compont::AllocV2()
197 : {
198 1 : inputMem = std::make_shared<DevBuffer>(dataCount * DataTypeSizeGet(DataType::INT64) * comm->GetRankSize());
199 3 : HCCL_INFO("[Mc2Compont][AllocV2]inputMem addr[%p] size = [%llu]", inputMem->GetAddr(), inputMem->GetSize());
200 9 : for(uint32_t i = 0; i < MAX_OP_NUM; i++) {
201 8 : combinOpParam.opType[i] = 0;
202 8 : combinOpParam.algorithmType[i] = 0;
203 : }
204 1 : if (ccuResourceAlloced) {
205 0 : return;
206 : }
207 :
208 1 : constexpr uint32_t comSyncNum = 2; // 每轮同步使用2个同步信号
209 1 : uint32_t comParamBufSize = CCU_TASK_NUM_MAX * CCU_PARAM_NUM_MAX * CCU_ONE_PARAM_SIZE ;
210 1 : uint32_t comSyncBufSize = CCU_TASK_NUM_MAX * comSyncNum * CCU_ONE_PARAM_SIZE ;
211 1 : workspaceBuffer = std::make_shared<DevBuffer>(MC2_WORKSPACE_SIZE);
212 1 : comParamBuffer = std::make_shared<DevBuffer>(comParamBufSize);
213 1 : comSyncBuffer = std::make_shared<DevBuffer>(comSyncBufSize);
214 1 : HrtMemset(reinterpret_cast<void*>(workspaceBuffer->GetAddr()), MC2_WORKSPACE_SIZE, MC2_WORKSPACE_SIZE);
215 1 : HrtMemset(reinterpret_cast<void*>(comParamBuffer->GetAddr()), comParamBufSize, comParamBufSize);
216 1 : HrtMemset(reinterpret_cast<void*>(comSyncBuffer->GetAddr()), comSyncBufSize, comSyncBufSize);
217 1 : combinOpParam.workSpace = static_cast<uint64_t>(workspaceBuffer->GetAddr());
218 1 : combinOpParam.workSpaceSize = MC2_WORKSPACE_SIZE;
219 1 : combinOpParam.rankId = comm->GetMyRank();
220 1 : combinOpParam.rankDim = comm->GetRankSize();
221 1 : combinOpParam.xnAddr = static_cast<uint64_t>(comParamBuffer->GetAddr());
222 1 : combinOpParam.ckeAddr = static_cast<uint64_t>(comSyncBuffer->GetAddr());
223 : // add cclbuffer info
224 1 : if (comm->GetCclBuffer() == nullptr) {
225 0 : THROW<Hccl::InternalException>(StringFormat("Cannot get CCL Buffer to fill window!"));
226 : }
227 1 : combinOpParam.winSize = static_cast<uint64_t>(comm->GetCclBuffer()->GetSize());
228 1 : combinOpParam.windowsOut[0] = static_cast<uint64_t>(comm->GetCclBuffer()->GetAddr());
229 1 : ccuResourceAlloced = true;
230 :
231 1 : tokenInfo = CcuRep::GetTokenInfo(static_cast<uint64_t>(workspaceBuffer->GetAddr()),
232 1 : static_cast<uint64_t>(workspaceBuffer->GetSize()));
233 : }
234 :
235 2 : void Mc2Compont::MC2Orchestrate(const CollAlgParams& params, std::shared_ptr<InsQueue>& insQueue, uint8_t commEngine) const
236 : {
237 2 : auto op = comm->GetCurrentCollOperator();
238 :
239 2 : CollOpParams opParams;
240 2 : opParams.commEngine = static_cast<HcclAccelerator::Value>(commEngine);
241 2 : opParams.opType = op->opType;
242 2 : opParams.dataType = op->dataType;
243 2 : opParams.count = op->dataCount;
244 2 : opParams.reduceOp = op->reduceOp;
245 2 : opParams.isMc2 = params.isMc2;
246 2 : comm->ExecAlgSelect(opParams, op->opMode);
247 2 : if (!comm->GetOpCcuFeatureFlag()) { // 算子粒度
248 0 : auto msg = StringFormat("[Mc2Compont:%s]AlgSelect not ccu, accState[%s]", __func__, comm->GetOpExecuteConfig().accState.Describe().c_str());
249 0 : THROW<InternalException>(msg);
250 0 : }
251 :
252 2 : std::string algName = comm->GetCurAlgName();
253 : // 算子编排获取InsQueue
254 2 : auto ret = comm->GetCollAlgComponent()->Orchestrate(*op, params, algName, insQueue);
255 2 : if (ret != HcclResult::HCCL_SUCCESS) {
256 1 : THROW<InternalException>(
257 3 : StringFormat("Error occurs when call collAlgComponent.orchestrate(), error code: %d", ret));
258 : }
259 3 : }
260 :
261 1 : void Mc2Compont::MC2AllocCommRes(const CollAlgParams ¶ms, std::shared_ptr<InsQueue> &insQueue, uint8_t commEngine) const
262 : {
263 1 : MC2Orchestrate(params, insQueue, commEngine);
264 : // 获取LinkData
265 1 : auto collService = dynamic_cast<CollServiceDeviceMode *>(comm->GetCollService());
266 1 : auto ccuLinks = collService->GetUniqueLinks(insQueue);
267 : // Socket建链
268 1 : comm->GetSocketManager().BatchCreateSockets(ccuLinks);
269 : // 对insQueue中ccuIns进行预处理(创建transport、ccuCtx、分配资源、注册等)
270 1 : collService->GetCcuInsPreprocessor()->Preprocess(insQueue, true);
271 1 : if (collService->GetCcuInsPreprocessor()->IsRollback()) { // mc2暂不能回退到aicpu
272 1 : THROW<InternalException>("[Mc2Compont][%s]ResAlloc unsuccessful.", __func__);
273 : }
274 1 : }
275 :
276 0 : void Mc2Compont::SaveAlgoInfo(uint32_t index, uint64_t templateSign, uint32_t opType, uint8_t algorithmType) {
277 0 : combinOpParam.opType[index] = opType;
278 0 : combinOpParam.algorithmType[index] = algorithmType;
279 0 : HcclAlgoInfo hcclAlgoInfo{};
280 0 : hcclAlgoInfo.opType = opType;
281 0 : hcclAlgoInfo.algorithmType = algorithmType;
282 0 : algoInfoMap_[templateSign] = hcclAlgoInfo;
283 0 : return;
284 : }
285 :
286 0 : void Mc2Compont::GenerateAlgoTemplates(Mc2Tiling *mc2TilingPtr, std::unordered_set<uint64_t> &algoTemplateRequire)
287 : {
288 0 : HCCL_INFO("GenerateAlgoTemplates start v1");
289 :
290 0 : auto tmpMemSize = comm->GetBufferSize();
291 0 : CollAlgParams params;
292 0 : params.opMode = OpMode::OPBASE;
293 0 : params.maxTmpMemSize = tmpMemSize;
294 0 : params.isMc2 = true;
295 : // 从mc2Tiling中获取需要的算法信息
296 0 : Mc2CommConfig *commConfigPtr = reinterpret_cast<Mc2CommConfig *>(
297 : reinterpret_cast<uint8_t *>(mc2TilingPtr) + sizeof(uint32_t) + sizeof(uint32_t) + sizeof(Mc2ServerCfg));
298 0 : for (uint32_t index = 0; index < mc2TilingPtr->commConfigNum; index++) {
299 0 : const auto &commConfig = *(commConfigPtr + index);
300 0 : OpParamsChecker::CheckOpDataTypeMC2(commConfig);
301 0 : uint64_t templateSign = GetTemplateSignature(commConfig);
302 0 : algoTemplateRequire.insert(templateSign);
303 : // 已经生成过的算法模板不再生成
304 0 : if (algoTemplateMap.find(templateSign) != algoTemplateMap.end()) {
305 0 : HCCL_INFO("A algoTemplate that meets the requirement already exists, index = [%u], templateSign = [%llu]", index, templateSign);
306 0 : if (algoInfoMap_.find(templateSign) != algoInfoMap_.end()) {
307 0 : combinOpParam.opType[index] = algoInfoMap_[templateSign].opType;
308 0 : combinOpParam.algorithmType[index] = algoInfoMap_[templateSign].algorithmType;
309 0 : continue;
310 : } else {
311 0 : THROW<Hccl::InternalException>(StringFormat("algoInfoMap_ do not has templateSign = [%llu]", templateSign));
312 : }
313 : }
314 :
315 0 : FillCollOperator(commConfig);
316 :
317 0 : auto insQueue = make_shared<InsQueue>();
318 0 : MC2AllocCommRes(params, insQueue, commConfig.communicationEngine);
319 :
320 0 : std::string algName = comm->GetCurAlgName();
321 0 : HCCL_INFO("Orchestrate: index = [%u], algName = [%s], templateSign = [%llu]", index, algName.c_str(), templateSign);
322 0 : if (insQueue->Iter()->GetType() != InstructionType::CCU_INS) {
323 0 : THROW<Hccl::InternalException>(StringFormat("InstructionType is not ccu ins, algName = [%s]", algName.c_str()));
324 : }
325 :
326 : // 获取taskParam
327 0 : const CcuInstruction& ccuInstruction = static_cast<const CcuInstruction &>(*insQueue->Iter());
328 0 : std::vector<std::vector<CcuTaskParam>> taskParams;
329 0 : ccuInstruction.Translate(taskParams);
330 0 : if (taskParams.empty()) {
331 0 : THROW<Hccl::InternalException>(StringFormat("CcuInstruction translate faild, index = [%u], algName = [%s]", index, algName.c_str()));
332 : }
333 0 : algoTemplateMap[templateSign] = taskParams;
334 0 : SaveAlgoInfo(index, templateSign, commConfig.opType, comm->GetAlgorithmType());
335 0 : for (const auto &task : taskParams) {
336 0 : HCCL_INFO("taskParam: dieId = [%u], instStartId = [%u]", task[0].dieId, task[0].instStartId);
337 0 : SaveMc2DfxTaskInfo(task[0], ccuInstruction.GetExecId());
338 : }
339 0 : }
340 0 : HCCL_INFO("GenerateAlgoTemplates success");
341 0 : }
342 :
343 1 : void Mc2Compont::GenerateAlgoTemplatesV2(const Mc2InitTilingInner *mc2TilingPtr, std::unordered_set<uint64_t> &algoTemplateRequire)
344 : {
345 3 : HCCL_INFO("GenerateAlgoTemplates start v2");
346 :
347 1 : auto tmpMemSize = comm->GetBufferSize();
348 1 : CollAlgParams params;
349 1 : params.opMode = OpMode::OPBASE;
350 1 : params.maxTmpMemSize = tmpMemSize;
351 1 : params.isMc2 = true;
352 1 : if(mc2TilingPtr->mc2HcommCnt > MAX_OP_NUM) {
353 0 : THROW<Hccl::InternalException>(StringFormat("mc2HcommCnt is lager than MAX_OP_NUM, mc2HcommCnt = [%u]", mc2TilingPtr->mc2HcommCnt));
354 : }
355 :
356 2 : for (uint32_t index = 0; index < mc2TilingPtr->mc2HcommCnt; index++) {
357 1 : const auto offset = mc2TilingPtr->offset[index];
358 1 : const auto &commConfig = *(reinterpret_cast<const Mc2CcTilingInner *>(reinterpret_cast<const uint8_t *>(mc2TilingPtr) + offset));
359 1 : OpParamsChecker::CheckOpDataTypeMC2V2(commConfig);
360 1 : uint64_t templateSign = GetTemplateSignatureV2(commConfig);
361 1 : algoTemplateRequire.insert(templateSign);
362 : // 已经生成过的算法模板不再生成
363 1 : if (algoTemplateMap.find(templateSign) != algoTemplateMap.end()) {
364 3 : HCCL_INFO("A algoTemplate that meets the requirement already exists, index = [%u], templateSign = [%llu]", index, templateSign);
365 1 : if (algoInfoMap_.find(templateSign) != algoInfoMap_.end()) {
366 1 : combinOpParam.opType[index] = algoInfoMap_[templateSign].opType;
367 1 : combinOpParam.algorithmType[index] = algoInfoMap_[templateSign].algorithmType;
368 1 : continue;
369 : } else {
370 0 : THROW<Hccl::InternalException>(StringFormat("algoInfoMap_ do not has templateSign = [%llu]", templateSign));
371 : }
372 : }
373 :
374 0 : FillCollOperatorV2(commConfig);
375 :
376 0 : auto insQueue = make_shared<InsQueue>();
377 0 : MC2AllocCommRes(params, insQueue, commConfig.communicationEngine);
378 :
379 0 : std::string algName = comm->GetCurAlgName();
380 0 : HCCL_INFO("Orchestrate: index = [%u], algName = [%s], templateSign = [%llu]", index, algName.c_str(), templateSign);
381 0 : if (insQueue->Iter()->GetType() != InstructionType::CCU_INS) {
382 0 : THROW<Hccl::InternalException>(StringFormat("InstructionType is not ccu ins, algName = [%s]", algName.c_str()));
383 : }
384 :
385 : // 获取taskParam
386 0 : const CcuInstruction& ccuInstruction = static_cast<const CcuInstruction &>(*insQueue->Iter());
387 0 : std::vector<std::vector<CcuTaskParam>> taskParams;
388 0 : ccuInstruction.Translate(taskParams);
389 0 : if (taskParams.empty()) {
390 0 : THROW<Hccl::InternalException>(StringFormat("CcuInstruction translate faild, index = [%u], algName = [%s]", index, algName.c_str()));
391 : }
392 0 : algoTemplateMap[templateSign] = taskParams;
393 0 : SaveAlgoInfo(index, templateSign, commConfig.opType, comm->GetAlgorithmType());
394 0 : for (const auto &task : taskParams) {
395 0 : HCCL_INFO("taskParam: dieId = [%u], instStartId = [%u]", task[0].dieId, task[0].instStartId);
396 0 : SaveMc2DfxTaskInfo(task[0], ccuInstruction.GetExecId());
397 : }
398 0 : }
399 3 : HCCL_INFO("GenerateAlgoTemplates success");
400 1 : }
401 :
402 3 : static std::map<uint8_t, std::map<uint32_t, uint32_t>> TransToMap(const std::vector<std::vector<CcuTaskParam>>& params)
403 : {
404 3 : std::map<uint8_t, std::map<uint32_t, uint32_t>> dieIdToInstrIdMap;
405 :
406 12 : for (const auto& param : params) {
407 9 : uint8_t dieId = param[0].dieId;
408 9 : uint32_t instrId = param[0].instStartId;
409 :
410 : // 检查 dieId 是否已经存在于 map 中
411 9 : if (dieIdToInstrIdMap.find(dieId) == dieIdToInstrIdMap.end()) {
412 : // 如果不存在，创建一个新的 map
413 5 : std::map<uint32_t, uint32_t> indexIdToInstrIdMap;
414 5 : indexIdToInstrIdMap[0] = instrId;
415 5 : dieIdToInstrIdMap[dieId] = indexIdToInstrIdMap;
416 5 : } else {
417 : // 如果存在，获取对应的 map
418 4 : std::map<uint32_t, uint32_t>& indexIdToInstrIdMap = dieIdToInstrIdMap[dieId];
419 : // 插入新的 IndexId 和 InstrId
420 4 : int indexId = indexIdToInstrIdMap.size();
421 4 : indexIdToInstrIdMap[indexId] = instrId;
422 : }
423 : }
424 3 : return dieIdToInstrIdMap;
425 0 : }
426 :
427 2 : static std::map<uint8_t, std::vector<uint32_t>> TransToDieIdMissionIdMap(const std::vector<std::vector<CcuTaskParam>>& params)
428 : {
429 2 : std::map<uint8_t, std::vector<uint32_t>> dieIdMissionIdMap;
430 :
431 7 : for (const auto& param : params) {
432 5 : uint8_t dieId = param[0].dieId;
433 5 : uint32_t missionId = param[0].missionId;
434 :
435 5 : if (dieIdMissionIdMap.find(dieId) == dieIdMissionIdMap.end()) {
436 3 : dieIdMissionIdMap[dieId] = std::vector<uint32_t>();
437 : }
438 5 : dieIdMissionIdMap[dieId].push_back(missionId);
439 : }
440 2 : return dieIdMissionIdMap;
441 0 : }
442 :
443 4 : bool Mc2Compont::CompareMissionMap(const std::map<uint8_t, std::map<uint32_t, uint32_t>> &mapA,
444 : const std::map<uint8_t, std::map<uint32_t, uint32_t>> &mapB) const
445 : {
446 4 : if (mapA.size() != mapB.size()) {
447 1 : return false;
448 : }
449 5 : for (auto &entry : mapA) {
450 4 : uint8_t curDieId = entry.first;
451 4 : if (mapB.find(curDieId) == mapB.end()) {
452 2 : return false;
453 : }
454 4 : const std::map<uint32_t, uint32_t> &curSubMapA = entry.second;
455 4 : const std::map<uint32_t, uint32_t> &curSubMapB = mapB.at(curDieId);
456 4 : if (curSubMapA.size() != curSubMapB.size()) {
457 2 : return false;
458 : }
459 6 : for (auto &elem : curSubMapA) {
460 4 : if (curSubMapB.find(elem.first) == curSubMapB.end()) {
461 0 : return false;
462 : }
463 : }
464 : }
465 1 : return true;
466 : }
467 :
468 5 : u32 Mc2Compont::GetCcuMc2ServerNum()
469 : {
470 5 : return ccuServerMap.size();
471 : }
472 :
473 3 : void Mc2Compont::GenerateCcuServer(const std::unordered_set<uint64_t> &algoTemplateRequire)
474 : {
475 9 : HCCL_INFO("GenerateCcuServer start");
476 3 : if (algoTemplateRequire.empty()) {
477 0 : THROW<InvalidParamsException>(StringFormat("AlgoTemplate require num is zero!"));
478 : }
479 :
480 : InsExeQue::ExtInsExeEntityId execId;
481 : // 查找当前是否存在符合条件的ccuServer
482 3 : if (FindCcuServer(algoTemplateRequire, execId)) {
483 1 : curExecId = execId;
484 3 : HCCL_INFO("A CcuServer that meets the requirement already exists, execId = [%llu]", execId);
485 1 : return;
486 : }
487 :
488 : // 没有符合条件的ccuServer, 生成一个新的ccuServer
489 2 : std::map<uint64_t, std::map<uint8_t, std::map<uint32_t, uint32_t>>> signatureMap;
490 2 : std::map<uint8_t, std::map<uint32_t, uint32_t>> compareMap;
491 2 : std::map<uint8_t, std::vector<uint32_t>> dieIdMissionIdMap;
492 2 : bool initFlag = false;
493 5 : for (uint64_t templateSignature : algoTemplateRequire) {
494 3 : auto tmpMap = TransToMap(algoTemplateMap[templateSignature]);
495 3 : if (!initFlag) {
496 2 : compareMap = tmpMap;
497 2 : dieIdMissionIdMap = TransToDieIdMissionIdMap(algoTemplateMap[templateSignature]);
498 2 : initFlag = true;
499 : } else {
500 1 : if (!CompareMissionMap(compareMap,tmpMap)) {
501 0 : THROW<InvalidParamsException>(StringFormat("AlgoTemplate require is not the same!"));
502 : }
503 : }
504 : // algoTemplateRequire为unordered_set，可以保证键值唯一
505 3 : signatureMap[templateSignature] = tmpMap;
506 3 : }
507 :
508 2 : std::map<uint8_t, std::map<uint32_t, std::map<uint64_t, uint32_t>>> algoTemplate;
509 5 : for (const auto& signature: signatureMap) {
510 8 : for (const auto& dieId: signature.second) {
511 14 : for (const auto& indexId: dieId.second) {
512 9 : algoTemplate[dieId.first][indexId.first][signature.first] = indexId.second;
513 : }
514 : }
515 : }
516 :
517 : // 实例化Mc2Context
518 2 : CcuCtxGroup ctxGroup;
519 2 : uint32_t dieNum = algoTemplate.size();
520 5 : for (const auto &item : algoTemplate) {
521 3 : uint8_t dieId = item.first;
522 8 : for (const auto &mission : item.second) {
523 5 : std::unique_ptr<Mc2ContextBase> mc2Context;
524 5 : if (mission.first == 0) {
525 3 : mc2Context = std::make_unique<Mc2Context>();
526 3 : mc2Context->SetDieId(dieId);
527 3 : static_cast<Mc2Context *>(mc2Context.get())->SetDieNum(dieNum);
528 3 : static_cast<Mc2Context *>(mc2Context.get())
529 3 : ->SetCommAddr(static_cast<uint64_t>(comSyncBuffer->GetAddr()),
530 3 : static_cast<uint64_t>(comParamBuffer->GetAddr()));
531 : } else {
532 2 : mc2Context = std::make_unique<Mc2SlaveContext>();
533 2 : mc2Context->SetDieId(dieId);
534 : }
535 5 : mc2Context->SetMissionNumAndId(item.second.size(), mission.first);
536 5 : mc2Context->SetAlgoTemplateInfo(mission.second);
537 5 : ctxGroup.ctxs.push_back(std::move(mc2Context));
538 5 : }
539 : }
540 :
541 : // 申请资源
542 2 : HcclResult ret = CcuCtxMgr::AllocRes(comm->GetDeviceLogicId(), ctxGroup, ccuResPack);
543 2 : if (ret != HcclResult::HCCL_SUCCESS) {
544 0 : THROW<CcuApiException>("GenerateCcuServer AllocRes failed, ret[%d]", ret);
545 : }
546 :
547 : // needtodo 检查各个templateSignature对应的missionId相同
548 2 : std::map<uint8_t, uint32_t> dieIndex;
549 7 : for (auto &ctx: ctxGroup.ctxs) {
550 5 : if (dieIndex.find(ctx->GetDieId()) == dieIndex.end()) {
551 3 : dieIndex[ctx->GetDieId()] = 0;
552 : }
553 5 : ctx->SetMissionId(dieIdMissionIdMap[ctx->GetDieId()][dieIndex[ctx->GetDieId()]++]);
554 : }
555 :
556 : // 指令注册
557 2 : InsExeQue::ExtInsExeEntity entity;
558 2 : entity.ctxGroup = std::move(ctxGroup);
559 2 : ret = InsExeQue::RegisterExtendInstruction(comm->GetDeviceLogicId(), entity, execId);
560 2 : if (ret != HcclResult::HCCL_SUCCESS) {
561 0 : THROW<CcuApiException>("GenerateCcuServer RegisterExtendInstruction failed, ret[%d]", ret);
562 : }
563 2 : ccuServerMap[execId] = algoTemplateRequire;
564 2 : curExecId = execId;
565 6 : HCCL_INFO("GenerateCcuServer success, execId[%llu]", execId);
566 2 : }
567 :
568 3 : bool Mc2Compont::FindCcuServer(const std::unordered_set<uint64_t> &algoTemplateRequire,
569 : InsExeQue::ExtInsExeEntityId &execId) const
570 : {
571 : // 查找是否有符合的ccuServer存在，本次算子需求是已有sever中包含算子的子集既符合要求
572 3 : for (const auto &server : ccuServerMap) {
573 1 : bool isMatch = true;
574 2 : for (auto &templateSign : algoTemplateRequire) {
575 1 : if (server.second.find(templateSign) == server.second.end()) {
576 0 : isMatch = false;
577 0 : break;
578 : }
579 : }
580 1 : if (isMatch) {
581 1 : execId = server.first;
582 1 : return true;
583 : }
584 : }
585 2 : return false;
586 : }
587 :
588 0 : uint64_t Mc2Compont::GetTemplateSignature(const Mc2CommConfig &config) const
589 : {
590 : // 根据Mc2CommConfig生成算子模板签名
591 0 : if (config.opType > UINT8_MAX || config.reduceType > UINT8_MAX || config.dataType > UINT8_MAX
592 0 : || config.outputDataType > UINT8_MAX) {
593 0 : THROW<InvalidParamsException>(
594 0 : StringFormat("MC2 High Level API GetTemplateSignature Failed, Mc2CommConfig value is bigger than 256!"));
595 : }
596 0 : constexpr uint16_t algoTypeShift = 32;
597 0 : constexpr uint16_t outputDataTypeShift = 24;
598 0 : constexpr uint16_t reduceTypeShift = 16;
599 0 : constexpr uint16_t dataTypeShift = 8;
600 0 : constexpr uint16_t opTypeShift = 0;
601 0 : uint64_t opType = config.opType;
602 0 : uint64_t dataType = config.dataType;
603 0 : uint64_t reduceType = config.reduceType;
604 0 : uint64_t outputDataType = config.outputDataType;
605 0 : uint64_t algoType = 0; // 用于算法选择，当前暂不支持，固定为0
606 :
607 0 : uint64_t templateSignature = ((opType & 0xff) << opTypeShift) | ((dataType & 0xff) << dataTypeShift)
608 0 : | ((reduceType & 0xff) << reduceTypeShift)
609 0 : | ((outputDataType & 0xff) << outputDataTypeShift)
610 0 : | ((algoType & 0xff) << algoTypeShift);
611 0 : HCCL_INFO("[GetTemplateSignature]: opType[%s] dataType[%s] reduceType[%s] outputDataType[%s] algoType[%u] "
612 : "templateSignature[%llu]",
613 : MC2OpType(static_cast<AicpuComType>(config.opType)).Describe().c_str(),
614 : MC2DataType(static_cast<HcclDataType>(config.dataType)).Describe().c_str(),
615 : MC2ReduceType(static_cast<HcclReduceOp>(config.reduceType)).Describe().c_str(),
616 : MC2DataType(static_cast<HcclDataType>(config.outputDataType)).Describe().c_str(), algoType,
617 : templateSignature);
618 0 : return templateSignature;
619 : }
620 :
621 2 : uint64_t Mc2Compont::GetTemplateSignatureV2(const Mc2CcTilingInner &config) const
622 : {
623 : // 根据Mc2CommConfig生成算子模板签名
624 2 : if (config.opType > UINT8_MAX || config.reduceType > UINT8_MAX) {
625 0 : THROW<InvalidParamsException>(
626 0 : StringFormat("MC2 High Level API GetTemplateSignature Failed, Mc2CommConfig value is bigger than 256!"));
627 : }
628 2 : constexpr uint16_t algoTypeShift = 32;
629 2 : constexpr uint16_t outputDataTypeShift = 24;
630 2 : constexpr uint16_t reduceTypeShift = 16;
631 2 : constexpr uint16_t dataTypeShift = 8;
632 2 : constexpr uint16_t opTypeShift = 0;
633 2 : uint64_t opType = config.opType;
634 2 : uint64_t dataType = config.srcDataType;
635 2 : uint64_t reduceType = config.reduceType;
636 2 : uint64_t outputDataType = config.dstDataType;
637 2 : uint64_t algoType = 0; // 用于算法选择，当前暂不支持，固定为0
638 :
639 2 : uint64_t templateSignature = ((opType & 0xff) << opTypeShift) | ((dataType & 0xff) << dataTypeShift)
640 2 : | ((reduceType & 0xff) << reduceTypeShift)
641 2 : | ((outputDataType & 0xff) << outputDataTypeShift)
642 2 : | ((algoType & 0xff) << algoTypeShift);
643 6 : HCCL_INFO("[GetTemplateSignature]: opType[%s] dataType[%s] reduceType[%s] outputDataType[%s] algoType[%u] "
644 : "templateSignature[%llu]",
645 : MC2OpType(static_cast<AicpuComType>(config.opType)).Describe().c_str(),
646 : MC2DataType(static_cast<HcclDataType>(config.srcDataType)).Describe().c_str(),
647 : MC2ReduceType(static_cast<HcclReduceOp>(config.reduceType)).Describe().c_str(),
648 : MC2DataType(static_cast<HcclDataType>(config.dstDataType)).Describe().c_str(), algoType,
649 : templateSignature);
650 2 : return templateSignature;
651 : }
652 :
653 1 : void Mc2Compont::FillCollOperatorV2(const Mc2CcTilingInner &config)
654 : {
655 1 : CollOpParams opParams;
656 1 : opParams.opType = MC2OpType(static_cast<AicpuComType>(config.opType));
657 1 : opParams.reduceOp = MC2ReduceType(static_cast<HcclReduceOp>(config.reduceType));
658 1 : opParams.dataType = MC2DataType(static_cast<HcclDataType>(config.srcDataType));
659 1 : opParams.outputDataType = MC2DataType(static_cast<HcclDataType>(config.dstDataType));
660 1 : opParams.count = dataCount;
661 1 : opParams.sendBuf = reinterpret_cast<void *>(inputMem->GetAddr());
662 1 : opParams.recvBuf = reinterpret_cast<void *>(inputMem->GetAddr());
663 1 : if (opParams.opType == OpType::ALLTOALL) {
664 0 : opParams.all2AllDataDes.sendType = opParams.dataType;
665 0 : opParams.all2AllDataDes.recvType = opParams.outputDataType;
666 0 : opParams.all2AllDataDes.sendCount = dataCount;
667 0 : opParams.all2AllDataDes.recvCount = dataCount;
668 : }
669 1 : std::string opTag = comm->GetId();
670 :
671 1 : if (opParams.opType == OpType::ALLTOALLV) {
672 1 : opParams.all2AllVDataDes.sendType = opParams.dataType;
673 1 : opParams.all2AllVDataDes.recvType = opParams.outputDataType;
674 1 : dataCounts.resize(comm->GetRankSize());
675 1 : displs.resize(comm->GetRankSize());
676 1 : u64 countSum = 0;
677 2 : for (u32 i = 0; i < comm->GetRankSize(); i++) {
678 1 : dataCounts.at(i) = 1;
679 1 : displs.at(i) = countSum++;
680 : }
681 1 : opParams.all2AllVDataDes.sendCounts = reinterpret_cast<void *>(&dataCounts[0]);
682 1 : opParams.all2AllVDataDes.recvCounts = reinterpret_cast<void *>(&dataCounts[0]);
683 1 : opParams.all2AllVDataDes.sdispls = reinterpret_cast<void *>(&displs[0]);
684 1 : opParams.all2AllVDataDes.rdispls = reinterpret_cast<void *>(&displs[0]);
685 : }
686 :
687 1 : comm->CovertToCurrentCollOperator(opTag, opParams, OpMode::OPBASE);
688 1 : }
689 :
690 1 : void Mc2Compont::FillCollOperator(const Mc2CommConfig &config)
691 : {
692 1 : CollOpParams opParams;
693 1 : opParams.opType = MC2OpType(static_cast<AicpuComType>(config.opType));
694 1 : opParams.reduceOp = MC2ReduceType(static_cast<HcclReduceOp>(config.reduceType));
695 1 : opParams.dataType = MC2DataType(static_cast<HcclDataType>(config.dataType));
696 1 : opParams.outputDataType = MC2DataType(static_cast<HcclDataType>(config.outputDataType));
697 1 : opParams.count = dataCount;
698 1 : opParams.sendBuf = reinterpret_cast<void *>(inputMem->GetAddr());
699 1 : opParams.recvBuf = reinterpret_cast<void *>(inputMem->GetAddr());
700 1 : if (opParams.opType == OpType::ALLTOALL) {
701 0 : opParams.all2AllDataDes.sendType = opParams.dataType;
702 0 : opParams.all2AllDataDes.recvType = opParams.outputDataType;
703 0 : opParams.all2AllDataDes.sendCount = dataCount;
704 0 : opParams.all2AllDataDes.recvCount = dataCount;
705 : }
706 1 : std::string opTag = comm->GetId();
707 :
708 1 : if (opParams.opType == OpType::ALLTOALLV) {
709 1 : opParams.all2AllVDataDes.sendType = opParams.dataType;
710 1 : opParams.all2AllVDataDes.recvType = opParams.outputDataType;
711 1 : dataCounts.resize(comm->GetRankSize());
712 1 : displs.resize(comm->GetRankSize());
713 1 : u64 countSum = 0;
714 2 : for (u32 i = 0; i < comm->GetRankSize(); i++) {
715 1 : dataCounts.at(i) = 1;
716 1 : displs.at(i) = countSum++;
717 : }
718 1 : opParams.all2AllVDataDes.sendCounts = reinterpret_cast<void *>(&dataCounts[0]);
719 1 : opParams.all2AllVDataDes.recvCounts = reinterpret_cast<void *>(&dataCounts[0]);
720 1 : opParams.all2AllVDataDes.sdispls = reinterpret_cast<void *>(&displs[0]);
721 1 : opParams.all2AllVDataDes.rdispls = reinterpret_cast<void *>(&displs[0]);
722 : }
723 :
724 1 : comm->CovertToCurrentCollOperator(opTag, opParams, OpMode::OPBASE);
725 1 : }
726 :
727 2 : void Mc2Compont::SaveMc2DfxTaskInfo(const CcuTaskParam& ccuTaskParam, uint64_t execId) const
728 : {
729 2 : shared_ptr<DfxOpInfo> dfxOpInfo = std::make_shared<DfxOpInfo>();
730 2 : dfxOpInfo->comm_ = comm;
731 :
732 2 : TaskParam taskParam{};
733 2 : taskParam.taskType = TaskParamType::TASK_CCU;
734 2 : taskParam.taskPara.Ccu.dieId = ccuTaskParam.dieId;
735 2 : taskParam.taskPara.Ccu.missionId = ccuTaskParam.missionId;
736 2 : taskParam.taskPara.Ccu.instrId = ccuTaskParam.instStartId;
737 2 : taskParam.taskPara.Ccu.executeId = execId;
738 :
739 2 : shared_ptr<TaskInfo> taskInfo = std::make_shared<TaskInfo>(0, 0, 0, taskParam, dfxOpInfo);
740 :
741 2 : MC2GlobalMirrorTasks::GetInstance().AddTaskInfo(comm->GetDeviceLogicId(), taskInfo);
742 2 : }
743 :
744 4 : std::vector<CcuTaskParam> Mc2Compont::GetAlgoCcuTaskInfo(InsExeQue::ExtInsExeEntityId execId) const
745 : {
746 4 : std::vector<CcuTaskParam> ccuTaskParam{};
747 4 : auto serverItor = ccuServerMap.find(execId);
748 4 : if (serverItor == ccuServerMap.end()) {
749 6 : HCCL_INFO("[Mc2Compont]Failed to find ccuServer by executeId[%llu]", execId);
750 2 : return ccuTaskParam;
751 : }
752 5 : for (uint64_t algoSign : serverItor->second) {
753 3 : auto algoTemplateItor = algoTemplateMap.find(algoSign);
754 3 : if (algoTemplateItor == algoTemplateMap.end()) {
755 3 : HCCL_INFO("[Mc2Compont]Failed to find ccuTaskParam by algoSign[%llu]", algoSign);
756 1 : continue;
757 1 : }
758 6 : for (const auto &taskParam : algoTemplateItor->second) {
759 4 : ccuTaskParam.push_back(taskParam[0]);
760 : }
761 : }
762 2 : return ccuTaskParam;
763 0 : }
764 : }
|
