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