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