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 "ccu_ins_preprocessor.h"
12 : #include "ccu_ctx_mgr.h"
13 : #include "ccu_ins_group.h"
14 : #include "ccu_transport.h"
15 : #include "ccu_communicator.h"
16 : #include "internal_exception.h"
17 : #include "ccu_transport_manager.h"
18 : #include "ccu_registered_ctx_mgr.h"
19 : #include "ccu_ctx_creator_registry.h"
20 : #include "ccu_transport_group_manager.h"
21 :
22 : namespace Hccl {
23 :
24 : constexpr u32 TEMP_MAX_CNTCKE_NUM = 16;
25 :
26 6 : static bool CreateCcuJettys(CcuCommunicator &ccuComm, const std::vector<LinkData> &links, bool &createStatus)
27 : {
28 6 : auto ret = ccuComm.GetCcuJettyMgr()->PrepareCreate(links);
29 6 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
30 1 : createStatus = false; // 预留处理资源不足回退情况，当前不支持回退
31 3 : HCCL_WARNING("[CcuInsPreprocessor][%s] create ccu jettys failed, "
32 : "ccu resource is unavaialble, please check.", __func__);
33 1 : return false;
34 : }
35 :
36 5 : if (ret != HcclResult::HCCL_SUCCESS) {
37 1 : createStatus = false;
38 3 : HCCL_ERROR("[CcuInsPreprocessor][%s] create ccu jettys failed, "
39 : "unexpected error, please check.", __func__);
40 1 : THROW<InternalException>("[CreateCcuJettys]create ccu jettys failed, unexpected error, please check.");
41 : }
42 :
43 4 : return true;
44 : }
45 :
46 4 : static bool CreateCcuTransports(CcuCommunicator &ccuComm, const std::vector<LinkData> &links, bool &createStatus,
47 : std::vector<CcuTransport *> &transports)
48 : {
49 4 : for (auto &link : links) {
50 2 : CcuTransport *tansport = nullptr;
51 2 : auto ret = ccuComm.GetCcuTransportMgr()->PrepareCreate(link, tansport);
52 2 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
53 1 : createStatus = false; // 预留处理资源不足回退情况，当前不支持回退
54 3 : HCCL_WARNING("[CcuInsPreprocessor][%s] create ccu transports failed, "
55 : "ccu resource is unavaialble, please check.", __func__);
56 1 : return false;
57 : }
58 :
59 1 : if (ret != HcclResult::HCCL_SUCCESS) {
60 1 : createStatus = false;
61 3 : HCCL_ERROR("[CcuInsPreprocessor][%s] create ccu transports failed, "
62 : "unexpected error, please check.", __func__);
63 1 : THROW<InternalException>("[CreateCcuTransports]create ccu transports failed, unexpected error, please check.");
64 : }
65 0 : transports.emplace_back(tansport);
66 : }
67 :
68 2 : return true;
69 : }
70 :
71 6 : std::unique_ptr<CcuContext> CcuInsPreprocessor::CreateCcuCtx(const CcuInstruction &ccuInst, bool &createStatus)
72 : {
73 18 : HCCL_INFO("[CcuInsPreprocessor::%s] start, ccuInst[%s].", __func__, ccuInst.Describe().c_str());
74 :
75 6 : std::vector<LinkData> links = ccuInst.GetLinks();
76 :
77 6 : if (!CreateCcuJettys(ccuComm, links, createStatus)) {
78 1 : return nullptr;
79 : }
80 :
81 4 : std::vector<CcuTransport *> transports;
82 4 : if (!CreateCcuTransports(ccuComm, links, createStatus, transports)) {
83 1 : return nullptr;
84 : }
85 :
86 : // 排序links, 使用RemoteRankId排序, 确保相同的通信模式每次生成的LinkGroup顺序一致, 避免重复创建TransportGroup
87 2 : std::sort(links.begin(), links.end(), [](const LinkData &a, const LinkData &b) {
88 0 : return a.GetRemoteRankId() < b.GetRemoteRankId();
89 : });
90 :
91 2 : vector<LinkInfo> linkInfos{};
92 2 : linkInfos.resize(links.size());
93 2 : std::transform(links.begin(), links.end(), linkInfos.begin(), [](const LinkData& link)
94 : {
95 0 : return LinkInfo{link};
96 : });
97 :
98 2 : LinkGroup linkGroup{linkInfos};
99 2 : u32 cntCkeNum = TEMP_MAX_CNTCKE_NUM; // 临时规避多轮不同算子导致CNTCKE资源不足，待后续正式方案修改
100 2 : CcuTransportGroup *transportGrp = ccuComm.GetCcuTransportGrpMgr()->PrepareCreate(linkGroup, cntCkeNum);
101 2 : if (transportGrp == nullptr) {
102 0 : createStatus = false; // transportGroup当前未适配资源不足场景，需重构
103 0 : HCCL_WARNING("[CcuInsPreprocessor::%s] transportGrp alloc resource fail, but fallback, "
104 : "transports size[%zu],rankGroup size[%zu], cntCkeNum[%u]",
105 : __func__, transports.size(), linkGroup.GetLinks().size(), cntCkeNum);
106 0 : return nullptr;
107 : }
108 :
109 6 : HCCL_INFO("[CcuInsPreprocessor::%s] create ccuContext end, transports size[%zu], createStatus[%d], "
110 : "rankGroup size[%zu], cntCkeNum[%u], createStatus[%d], ccuInst[%s].",
111 : __func__, transports.size(), createStatus, linkGroup.GetLinks().size(), cntCkeNum, createStatus,
112 : ccuInst.GetInstType().Describe().c_str());
113 :
114 2 : std::unique_ptr<CcuCtxArg> ctxArg = ccuInst.GetCtxArg();
115 2 : CHECK_NULLPTR(ctxArg, "[CcuInsPreprocessor::CreateCcuCtx] ctxArg is nullptr!");
116 4 : return CcuCtxCreatorRegistry::GetInstance().GetCreateFunc(ccuInst.GetInstType())(*ctxArg, transports,
117 2 : *transportGrp);
118 7 : }
119 :
120 2 : void CcuInsPreprocessor::CreateCcuCtxGroup(const CcuInstruction &ccuIns, std::unique_ptr<CcuCtxGroup> &ccuCtxGroupPtr,
121 : bool &createStatus)
122 : {
123 6 : HCCL_INFO("[CcuInsPreprocessor::%s] start, ccuIns[%s].", __func__, ccuIns.Describe().c_str());
124 :
125 2 : CcuInstType insType = ccuIns.GetInstType();
126 2 : if (insType == CcuInstType::CCU_INS_GROUP) {
127 0 : const CcuInsGroup &ccuInsGroup = dynamic_cast<const CcuInsGroup &>(ccuIns);
128 0 : for (auto &ins : ccuInsGroup.GetCcuInstructions()) {
129 0 : std::unique_ptr<CcuContext> ctxPtr = CreateCcuCtx(*ins, createStatus);
130 0 : if (ctxPtr == nullptr) {
131 0 : return;
132 : }
133 0 : ccuCtxGroupPtr->ctxs.emplace_back(std::move(ctxPtr));
134 0 : }
135 : } else {
136 2 : std::unique_ptr<CcuContext> ctxPtr = CreateCcuCtx(ccuIns, createStatus);
137 2 : if (ctxPtr == nullptr) {
138 0 : return;
139 : }
140 2 : ccuCtxGroupPtr->ctxs.emplace_back(std::move(ctxPtr));
141 2 : }
142 :
143 6 : HCCL_INFO("[CcuInsPreprocessor::%s] create ccuCtx end, insType[%s], ccuCtxGroup ctxs size[%zu], "
144 : "createStatus[%d].",
145 : __func__, insType.Describe().c_str(), ccuCtxGroupPtr->ctxs.size(), createStatus);
146 : }
147 :
148 4 : bool CcuInsPreprocessor::CheckCtxTransportStatus(bool resAllocSuccess)
149 : {
150 4 : if (!resAllocSuccess) {
151 3 : HCCL_INFO("[CcuInsPreprocessor::%s] CreateCcuCtx alloc local resource fail, ccuCtxGroups"
152 : " size[%zu], resPackIdxs size[%zu], ctxSignatures size[%zu], insPtrs size[%zu]",
153 : __func__, ccuCtxGroups.size(), resPackIdxs.size(), ctxSignatures.size(), insPtrs.size());
154 1 : return false;
155 : }
156 :
157 9 : HCCL_INFO("[CcuInsPreprocessor::%s] CreateCcuCtx success, ccuCtxGroups size[%zu], resPackIdxs"
158 : " size[%zu], ctxSignatures size[%zu], insPtrs size[%zu]",
159 : __func__, ccuCtxGroups.size(), resPackIdxs.size(), ctxSignatures.size(), insPtrs.size());
160 3 : return true;
161 : }
162 :
163 5 : void CcuInsPreprocessor::InsPreprocess(InsIterator &insIter, u32 resPackIndex, bool isMc2)
164 : {
165 5 : CcuInstruction &ccuIns = dynamic_cast<CcuInstruction &>(*insIter);
166 : // 获取Signature
167 5 : CcuCtxSignature ctxSignature = ccuIns.GetCtxSignature();
168 5 : if (isMc2) {
169 0 : ctxSignature.Append("_Mc2");
170 : }
171 : // 获取ResPack
172 5 : CcuResPack &ccuResPack = ccuComm.GetCcuResPackMgr()->GetCcuResPack(resPackIndex);
173 5 : uintptr_t resPackId = ccuResPack.GetId();
174 :
175 5 : u64 execId = 0;
176 5 : if (!ccuComm.GetRegisteredCcuCtxMgr()->HasRegistered(ctxSignature, resPackId, execId)) {
177 4 : needHandShake = true;
178 :
179 4 : resPackIdxs.emplace_back(resPackIndex);
180 4 : insPtrs.emplace_back(insIter);
181 4 : ctxSignatures.emplace_back(ctxSignature);
182 4 : if ((ccuCtxGroups.find(ctxSignature) != ccuCtxGroups.end())
183 4 : && (ccuCtxGroups[ctxSignature].find(resPackIndex) != ccuCtxGroups[ctxSignature].end())) {
184 1 : return;
185 : }
186 :
187 : // 根据CCU扩展指令创建CcuContext实例
188 4 : std::unique_ptr<CcuCtxGroup> ccuCtxGroupPtr = std::make_unique<CcuCtxGroup>();
189 4 : CHECK_NULLPTR(ccuCtxGroupPtr, "[CcuInsPreprocessor::InsPreprocess] ccuCtxGroupPtr is nullptr!");
190 4 : bool createStatus = true;
191 4 : CreateCcuCtxGroup(ccuIns, ccuCtxGroupPtr, createStatus);
192 :
193 : // 保存一些中间信息，用于后续的注册回退
194 4 : ccuCtxGroups[ctxSignature][resPackIndex] = std::move(ccuCtxGroupPtr);
195 4 : resAllocSuccess = createStatus && resAllocSuccess;
196 4 : if (!CheckCtxTransportStatus(resAllocSuccess)) {
197 1 : return;
198 : }
199 :
200 : // 调用平台层接口，为CcuContext实例分配本地CCU资源
201 : HcclResult res
202 3 : = CcuCtxMgr::AllocRes(ccuComm.GetDeviceLogicId(), *(ccuCtxGroups[ctxSignature][resPackIndex]), ccuResPack);
203 3 : if (res != HcclResult::HCCL_SUCCESS) {
204 1 : if (res != HcclResult::HCCL_E_UNAVAIL) {
205 1 : THROW<InternalException>("[CcuCtxMgr::AllocRes]AllocRes failed, unexpected error, please check.");
206 : }
207 0 : HCCL_INFO("[CcuInsPreprocessor::%s] AllocRes failed, ret[%u]", __func__, res);
208 0 : resAllocSuccess = false;
209 : }
210 4 : } else {
211 1 : ccuIns.SetExecId(execId);
212 : }
213 :
214 9 : HCCL_INFO("[CcuInsPreprocessor::%s] end, ccuResPack handles size[%zu].", __func__, ccuResPack.handles.size());
215 5 : }
216 :
217 7 : void CcuInsPreprocessor::PrepareCcuCtx(std::shared_ptr<InsQueue> &insQueue, bool isMc2)
218 : {
219 21 : HCCL_INFO("[CcuInsPreprocessor::%s] start, isMc2[%d].", __func__, isMc2);
220 :
221 : // 对每个从队列分配一个ResPack，对每个queue中每个ins进行预处理(构造ctx且分配本地资源)
222 7 : u32 resPackIndex = 0;
223 10 : for (auto slaveIter = insQueue->UnConstIterSlaves(); slaveIter.HasNext(); ++slaveIter) {
224 9 : for (auto ins = slaveIter->UnConstIter(); ins.HasNext(); ++ins) {
225 6 : if (ins->GetType() != InstructionType::CCU_INS) {
226 6 : HCCL_INFO("[CcuInsPreprocessor::%s] slave insQueue ins type[%s] not ccu type.", __func__,
227 : ins->GetType().Describe().c_str());
228 2 : continue;
229 2 : }
230 4 : InsPreprocess(ins, resPackIndex, isMc2);
231 4 : if (needHandShake && !resAllocSuccess) {
232 0 : HCCL_INFO("[CcuInsPreprocessor::%s] slave insQueue ins alloc local resource fail, "
233 : "resPackIndex[%u], ins[%s].",
234 : __func__, resPackIndex, ins->Describe().c_str());
235 0 : return;
236 : }
237 3 : }
238 3 : resPackIndex++;
239 7 : }
240 :
241 : // 主队列分配一个ResPack对每个ins进行预处理(构造ctx且分配本地资源)
242 13 : for (auto ins = insQueue->UnConstIter(); ins.HasNext(); ++ins) {
243 6 : if (ins->GetType() != InstructionType::CCU_INS) {
244 6 : HCCL_INFO("[CcuInsPreprocessor::%s] master insQueue ins type[%s] not ccu type.", __func__,
245 : ins->GetType().Describe().c_str());
246 2 : continue;
247 2 : }
248 4 : InsPreprocess(ins, resPackIndex, isMc2);
249 4 : if (needHandShake && !resAllocSuccess) {
250 0 : HCCL_INFO("[CcuInsPreprocessor::%s] master insQueue ins alloc local resource fail, "
251 : "resPackIndex[%u], ins[%s].",
252 : __func__, resPackIndex, ins->Describe().c_str());
253 0 : return;
254 : }
255 7 : }
256 :
257 21 : HCCL_INFO("[CcuInsPreprocessor::%s] prepare ccuContext end, needHandShake[%d], resAllocSuccess[%d].", __func__,
258 : needHandShake, resAllocSuccess);
259 : }
260 :
261 3 : void CcuInsPreprocessor::Confirm()
262 : {
263 9 : HCCL_INFO("[CcuInsPreprocessor::%s] start.", __func__);
264 :
265 : // 建链并交换
266 3 : ccuComm.GetCcuResPackMgr()->Confirm();
267 3 : ccuComm.GetCcuTransportMgr()->Confirm();
268 3 : ccuComm.GetCcuTransportGrpMgr()->Confirm();
269 3 : ccuComm.GetCcuJettyMgr()->Confirm();
270 :
271 9 : HCCL_INFO("[CcuInsPreprocessor::%s] confirm resource end.", __func__);
272 3 : }
273 :
274 5 : void CcuInsPreprocessor::Fallback()
275 : {
276 15 : HCCL_INFO("[CcuInsPreprocessor::%s] start.", __func__);
277 :
278 : // ccu回退流程resPack.handles为空，不需要删除resPack.handles
279 5 : ccuComm.GetCcuResPackMgr()->Fallback();
280 5 : ccuComm.GetCcuTransportMgr()->Fallback();
281 5 : ccuComm.GetCcuTransportGrpMgr()->Fallback();
282 5 : ccuComm.GetCcuJettyMgr()->Fallback();
283 :
284 15 : HCCL_INFO("[CcuInsPreprocessor::%s] fallback resource end.", __func__);
285 5 : }
286 :
287 3 : void CcuInsPreprocessor::RegisterCtx(bool isFuncBlock)
288 : {
289 9 : HCCL_INFO("[CcuInsPreprocessor::%s] start, isFuncBlock[%d].", __func__, isFuncBlock);
290 :
291 3 : u32 size = insPtrs.size();
292 3 : u64 execId = 0;
293 4 : for (u32 index = 0; index < size; ++index) {
294 : #ifdef HCCL_ALG_ANALYZER_DAVID
295 : // 为了算法分析器获取CCU微码序列，需要使用全局变量记录CCU指令和CCU微码，并建立映射关系
296 : extern Hccl::Instruction* g_ccuIns;
297 : g_ccuIns = &(*(insPtrs[index]));
298 : #endif
299 1 : uintptr_t resPackId = ccuComm.GetCcuResPackMgr()->GetCcuResPack(resPackIdxs[index]).GetId();
300 1 : CcuCtxSignature signature = ctxSignatures[index];
301 1 : if (!ccuComm.GetRegisteredCcuCtxMgr()->HasRegistered(signature, resPackId, execId)) {
302 1 : execId = ccuComm.GetRegisteredCcuCtxMgr()->Register(std::move(ccuCtxGroups[signature][resPackIdxs[index]]),
303 : signature, resPackId, isFuncBlock);
304 : }
305 1 : CcuInstruction &ccuIns = dynamic_cast<CcuInstruction &>(*insPtrs[index]);
306 1 : ccuIns.SetExecId(execId);
307 1 : }
308 :
309 9 : HCCL_INFO("[CcuInsPreprocessor::%s] register ctx end, insPtrs size[%u].", __func__, size);
310 3 : }
311 :
312 6 : void CcuInsPreprocessor::ClearTmpResRecords()
313 : {
314 6 : ccuCtxGroups.clear();
315 6 : resPackIdxs.clear();
316 6 : ctxSignatures.clear();
317 6 : insPtrs.clear();
318 6 : needHandShake = false;
319 6 : resAllocSuccess = true;
320 6 : }
321 :
322 7 : void CcuInsPreprocessor::Preprocess(std::shared_ptr<InsQueue> &insQueue, bool isMc2)
323 : {
324 21 : HCCL_INFO("[CcuInsPreprocessor::%s] insQueue Preprocess start.", __func__);
325 7 : isRollback = false;
326 : // ccuResPack资源扩充
327 7 : u32 insQueueSize = insQueue->SizeOfSlaves() + 1; // 从流个数 + 主流
328 7 : ccuComm.GetCcuResPackMgr()->PrepareAlloc(insQueueSize);
329 :
330 : // 对insQ中每个ins,创建CcuContext实例且分配资源
331 7 : PrepareCcuCtx(insQueue, isMc2);
332 7 : bool isFuncBlock = isMc2; // mc2场景需要将ctxGroup注册为FuncBlock
333 :
334 21 : HCCL_INFO("[CcuInsPreprocessor::%s] resAllocSuccess is[%d]", __func__, resAllocSuccess);
335 7 : if (!resAllocSuccess) {
336 12 : HCCL_INFO("[CcuInsPreprocessor::%s] ResAlloc unsuccessful, accelerator fall back.", __func__);
337 4 : if (isMc2) {
338 : // mc2场景，CCU资源不足时不支持回退
339 2 : THROW<InternalException>(StringFormat("[CcuInsPreprocessor::%s] Alloc local resource failed", __func__));
340 : }
341 : // CCU资源不足时warning
342 9 : HCCL_WARNING("[CcuInsPreprocessor::%s] Alloc local resource failed", __func__);
343 : // 若本地资源申请失败,且非用户显式配置CCU模式，则进行握手回退
344 3 : Fallback();
345 3 : ClearTmpResRecords();
346 3 : ccuComm.AcceleratorFallback();
347 1 : isRollback = true;
348 1 : return;
349 : }
350 : // 若本地资源申请成功, 则进行握手
351 : // 资源确认
352 3 : TRY_CATCH_PROCESS_THROW (
353 : InternalException,
354 : Confirm(),
355 : "[CCU Confirm] Comfirm Resources Error",
356 : // 建链失败时，清除临时创建的资源
357 : ClearTmpResRecords()
358 : );
359 :
360 : // 注册
361 3 : RegisterCtx(isFuncBlock);
362 3 : ClearTmpResRecords();
363 9 : HCCL_INFO("[CcuInsPreprocessor::%s] insQueue Preprocess end.", __func__);
364 : }
365 :
366 44 : CcuCommunicator *CcuInsPreprocessor::GetCcuComm()
367 : {
368 44 : return &ccuComm;
369 : }
370 :
371 261 : CcuInsPreprocessor::~CcuInsPreprocessor()
372 : {
373 261 : ccuCtxGroups.clear();
374 261 : resPackIdxs.clear();
375 261 : ctxSignatures.clear();
376 261 : insPtrs.clear();
377 261 : }
378 :
379 1 : HcclResult CcuInsPreprocessor::RecoverCcuTransportCtx(const std::vector<LinkData> &links,
380 : vector<std::pair<LinkGroup, u32>> linkGroupPair)
381 : {
382 3 : HCCL_INFO("[CcuInsPreprocessor::%s] start, links size[%u], linkGroupPair size[%u]", __func__, links.size(),
383 : linkGroupPair.size());
384 :
385 4 : CHK_RET(ccuComm.GetCcuJettyMgr()->PrepareCreate(links));
386 :
387 0 : std::vector<CcuTransport *> transports;
388 0 : transports.reserve(links.size());
389 0 : for (auto &link : links) {
390 0 : CcuTransport *tansport = nullptr;
391 0 : CHK_RET(ccuComm.GetCcuTransportMgr()->PrepareCreate(link, tansport));
392 0 : transports.emplace_back(tansport);
393 : }
394 :
395 : // u32 cntCkeNum = TEMP_MAX_CNTCKE_NUM; // 临时规避多轮不同算子导致CNTCKE资源不足，待后续正式方案修改
396 0 : for (auto &iter : linkGroupPair) {
397 0 : LinkGroup linkGroup = iter.first;
398 0 : u32 cntCkeNum = iter.second;
399 0 : CcuTransportGroup *transportGrp = ccuComm.GetCcuTransportGrpMgr()->PrepareCreate(linkGroup, cntCkeNum);
400 0 : if (transportGrp == nullptr) {
401 0 : HCCL_ERROR("[CcuInsPreprocessor::%s] transportGrp alloc resource fail, but fallback, "
402 : "transports size[%zu],rankGroup size[%zu], cntCkeNum[%u]",
403 : __func__, transports.size(), linkGroup.GetLinks().size(), cntCkeNum);
404 0 : return HCCL_E_INTERNAL;
405 : }
406 0 : }
407 0 : HCCL_INFO("[CcuInsPreprocessor::%s] end.", __func__);
408 0 : return HCCL_SUCCESS;
409 0 : }
410 :
411 1 : HcclResult CcuInsPreprocessor::RecoverCcuTransportConfirm()
412 : {
413 3 : HCCL_INFO("[CcuInsPreprocessor::%s] start.", __func__);
414 :
415 : // 建链并交换
416 1 : ccuComm.GetCcuTransportMgr()->RecoverConfirm();
417 1 : ccuComm.GetCcuTransportGrpMgr()->Confirm();
418 1 : ccuComm.GetCcuJettyMgr()->Confirm();
419 :
420 3 : HCCL_INFO("[CcuInsPreprocessor::%s] confirm resource end.", __func__);
421 1 : return HCCL_SUCCESS;
422 : }
423 :
424 3 : bool CcuInsPreprocessor::IsRollback() const
425 : {
426 3 : return isRollback;
427 : }
428 :
429 : } // namespace Hccl
|
