Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "coll_service_device_mode.h"
12 : #include "exception_util.h"
13 : #include "communicator_impl.h"
14 : #include "ccu_dev_mgr.h"
15 : #include "types.h"
16 : #include "aiv_ins.h"
17 : #include "stream_utils.h"
18 : #include "orion_adapter_rts.h"
19 :
20 : namespace Hccl {
21 :
22 : constexpr u32 SIZE_TABLE_ORION[HCCL_DATA_TYPE_RESERVED] = {sizeof(s8), sizeof(s16), sizeof(s32),
23 : 2, sizeof(float), sizeof(s64), sizeof(u64), sizeof(u8), sizeof(u16), sizeof(u32),
24 : 8, 2, 16, 2, 1, 1, 1, 1};
25 :
26 193 : void CollServiceDeviceMode::Init()
27 : {
28 579 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
29 : // 算子执行次数计数器buffer申请
30 193 : AddOpCounterMems();
31 579 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
32 193 : }
33 :
34 5 : static void AddCcuInsAndAicpuInsLinks(std::vector<LinkData> &linkDatas, const Instruction &ins)
35 : {
36 5 : InstructionType insType = ins.GetType();
37 5 : std::vector<LinkData> tmpLinkDatas;
38 5 : if (insType == InstructionType::CCU_INS) {
39 2 : const CcuInstruction &ccuIns = dynamic_cast<const CcuInstruction &>(ins);
40 2 : tmpLinkDatas = ccuIns.GetLinks();
41 3 : } else if (insType == InstructionType::AICPU_INS) {
42 2 : const AicpuInstruction &aicpuIns = dynamic_cast<const AicpuInstruction &>(ins);
43 2 : tmpLinkDatas = aicpuIns.GetLinks();
44 1 : } else if (insType == InstructionType::AIV_INS) {
45 1 : const AivInstruction &aivIns = dynamic_cast<const AivInstruction &>(ins);
46 1 : tmpLinkDatas = aivIns.GetLinks();
47 : }
48 5 : linkDatas.insert(linkDatas.end(), tmpLinkDatas.begin(), tmpLinkDatas.end());
49 5 : }
50 :
51 12 : std::vector<LinkData> CollServiceDeviceMode::GetUniqueLinks(std::shared_ptr<InsQueue> &insQueue) const
52 : {
53 36 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
54 :
55 : // 返回队列中所有ins的links
56 12 : std::vector<LinkData> links;
57 13 : for (auto slaveIter = insQueue->IterSlaves(); slaveIter.HasNext(); ++slaveIter) {
58 3 : for (auto ins = slaveIter->Iter(); ins.HasNext(); ++ins) {
59 2 : AddCcuInsAndAicpuInsLinks(links, *ins);
60 1 : }
61 12 : }
62 :
63 36 : HCCL_INFO("[CollServiceDeviceMode::%s] slaves end.", __func__);
64 :
65 15 : for (auto ins = insQueue->Iter(); ins.HasNext(); ++ins) {
66 3 : AddCcuInsAndAicpuInsLinks(links, *ins);
67 12 : }
68 :
69 12 : std::unordered_set<LinkData> linkDataSet(links.begin(), links.end());
70 12 : links.assign(linkDataSet.begin(), linkDataSet.end());
71 :
72 36 : HCCL_INFO("[CollServiceDeviceMode::%s] end, links size[%zu]", __func__, links.size());
73 12 : return links;
74 12 : }
75 :
76 7 : void CollServiceDeviceMode::LoadWithOpBasedMode(CollOperator &op, std::unique_ptr<Stream> stream)
77 : {
78 21 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
79 : // AIV aclgrah 流程
80 7 : if (comm->GetOpExecuteConfig().accState == AcceleratorState::AIV
81 7 : || comm->GetOpExecuteConfig().accState == AcceleratorState::AIV_ONLY) {
82 0 : HandleAclGraphFirstOpAivBuff(stream->GetPtr());
83 : }
84 :
85 : // 入参buffer和stream注册
86 7 : RegisterOpBufToBufMgr(op);
87 :
88 7 : RegisterOpbasedStream(std::move(stream));
89 :
90 7 : if (comm->GetOpExecuteConfig().accState == AcceleratorState::AIV
91 7 : || comm->GetOpExecuteConfig().accState == AcceleratorState::AIV_ONLY) {
92 0 : auto insQueue = make_shared<InsQueue>();
93 :
94 0 : AivOpCacheArgs opCacheParam{comm->GetCurAlgName(), op.dataCount, op.dataType, op.opType, op.reduceOp, op.root, op.numBlocksLimit, op.outputDataType,{},{}};
95 0 : if(op.opType == OpType::ALLTOALL){
96 0 : opCacheParam.all2allDataDes = {op.all2AllDataDes.sendType, op.all2AllDataDes.recvType, op.all2AllDataDes.sendCount, op.all2AllDataDes.recvCount};
97 : }
98 0 : if(op.opType == OpType::ALLTOALLV){
99 0 : opCacheParam.all2allVDataDes = {op.all2AllVDataDes.sendType, op.all2AllVDataDes.recvType, op.all2AllVDataDes.sendCounts, op.all2AllVDataDes.recvCounts,
100 0 : op.all2AllVDataDes.sdispls,op.all2AllVDataDes.rdispls};
101 : }
102 0 : auto it = comm->hcclCacheMap_.find(opCacheParam);
103 0 : bool isCache = false;
104 0 : bool isSendRecv = ((op.opType == OpType::SEND) || (op.opType == OpType::RECV) || (op.opType == OpType::BATCHSENDRECV));
105 0 : bool isAlltoAllV = (op.opType == OpType::ALLTOALLV);
106 0 : if ((it != comm->hcclCacheMap_.end()) && (!isSendRecv) && (!isAlltoAllV)) {
107 0 : isCache = true;
108 0 : insQueue = it->second;
109 : } else{
110 : // 算法编排返回insQueue, 包含ccu扩展指令和aicpu扩展指令
111 0 : insQueue = Orchestrate(op);
112 : }
113 0 : AllocQueueNotify(*insQueue);
114 : // 日志打印
115 0 : if (comm->GetAivTag() == 1) {
116 0 : std::vector<LinkData> uniqueLinks = comm->GetFullMeshLinks();
117 0 : comm->SetCommStatus(CommStatus::COMM_BUILDING);
118 : // Socket建链
119 0 : comm->GetSocketManager().BatchCreateSockets(uniqueLinks);
120 0 : aivInsPreprocessor.Preprocess(insQueue);
121 0 : }
122 : // translate
123 0 : SaveMirrorDfxOpInfo();
124 0 : Interpreter interpreter(*comm);
125 0 : interpreter.Submit(*insQueue);
126 0 : if(!isCache){
127 0 : comm->GetCacheMap(opCacheParam, insQueue);
128 : }
129 0 : } else {
130 : // 用于aicpu专用流
131 7 : comm->GetAicpuStreamManager().AllocFreeStream();
132 : // 算法编排返回insQueue, 包含ccu扩展指令和aicpu扩展指令
133 7 : shared_ptr<InsQueue> insQueue = Orchestrate(op);
134 7 : AllocQueueNotify(*insQueue);
135 : // 获取insQueue中所有Ins的linkDats
136 7 : std::vector<LinkData> uniqueLinks = GetUniqueLinks(insQueue);
137 : // 将通讯域设置为transport建链中状态
138 7 : comm->SetCommStatus(CommStatus::COMM_BUILDING);
139 :
140 : // Socket建链
141 7 : comm->GetSocketManager().BatchCreateSockets(uniqueLinks);
142 :
143 : // 对insQueue中ccuIns进行预处理(transport建链和交换, 资源申请、注册等)
144 7 : ccuInsPreprocessor.Preprocess(insQueue);
145 :
146 6 : if (ccuInsPreprocessor.IsRollback()) { // 如果是回退，流程退出
147 0 : return;
148 : }
149 6 : SaveMirrorDfxOpInfo();
150 : // translate
151 6 : Interpreter interpreter(*comm);
152 6 : interpreter.Submit(*insQueue);
153 8 : }
154 18 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
155 : }
156 :
157 2 : void CollServiceDeviceMode::LoadWithOffloadMode(CollOperator &op, std::unique_ptr<Stream> stream)
158 : {
159 6 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
160 :
161 2 : RegisterOpBufToBufMgr(op);
162 :
163 2 : RegisterOffloadMasterStream(op.opTag, std::move(stream));
164 :
165 : // 算法编排返回insQueue, 包含ccu扩展指令和aicpu扩展指令
166 2 : shared_ptr<InsQueue> insQueue = Orchestrate(op);
167 :
168 2 : AllocQueueNotify(*insQueue);
169 :
170 : // 获取insQueue中所有Ins的linkDats
171 2 : std::vector<LinkData> uniqueLinks = GetUniqueLinks(insQueue);
172 :
173 : // 将通讯域设置为transport建链中状态
174 2 : comm->SetCommStatus(CommStatus::COMM_BUILDING);
175 :
176 : // Socket建链
177 2 : comm->GetSocketManager().BatchCreateSockets(uniqueLinks);
178 :
179 : // 对insQueue中ccuIns进行预处理(transport建链和交换, 资源申请、注册等)
180 2 : aivInsPreprocessor.Preprocess(insQueue);
181 2 : ccuInsPreprocessor.Preprocess(insQueue);
182 :
183 1 : if (ccuInsPreprocessor.IsRollback()) { // 如果是回退，流程退出
184 0 : return;
185 : }
186 :
187 1 : SaveMirrorDfxOpInfo();
188 :
189 : // 下发head算子执行计数器task
190 1 : AddCountTask(true);
191 :
192 : // translate
193 1 : Interpreter interpreter(*comm);
194 1 : interpreter.Submit(*insQueue);
195 :
196 : // 下发tail算子执行计数器task
197 1 : AddCountTask(false);
198 :
199 3 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
200 3 : }
201 :
202 10 : shared_ptr<InsQueue> CollServiceDeviceMode::Orchestrate(const CollAlgOperator &op) const
203 : {
204 30 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
205 10 : u64 tmpMemSize = 0;
206 13 : if (op.opMode == OpMode::OPBASE || comm->GetOpExecuteConfig().accState == AcceleratorState::AIV
207 13 : || comm->GetOpExecuteConfig().accState == AcceleratorState::AIV_ONLY) {
208 8 : tmpMemSize = comm->GetBufferSize();
209 2 : } else if (op.scratchMem != nullptr) {
210 2 : tmpMemSize = op.scratchMem->GetSize();
211 : } else {
212 0 : HCCL_WARNING("[CollServiceDeviceMode::%s] no need scratchMem.", __func__);
213 : }
214 10 : CollAlgParams params;
215 10 : auto insQueue = make_shared<InsQueue>();
216 :
217 10 : params.opMode = op.opMode;
218 10 : params.maxTmpMemSize = tmpMemSize;
219 30 : HCCL_INFO("[CollServiceDeviceMode::%s] opMode[%d], tmpMemSize[%llu]", __func__, op.opMode, tmpMemSize);
220 10 : HcclResult errCode = comm->GetCollAlgComponent()->Orchestrate(op, params,comm->GetCurAlgName(), insQueue);
221 30 : HCCL_INFO("[CollServiceDeviceMode::%s] orchestrate with Ins end", __func__);
222 :
223 10 : if (errCode != HcclResult::HCCL_SUCCESS) {
224 0 : auto msg = StringFormat("Error occurs when call collAlgComponent.orchestrate(), error code: %d", errCode);
225 0 : THROW<InternalException>(msg);
226 0 : }
227 :
228 30 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
229 10 : return insQueue;
230 10 : }
231 :
232 1 : void CollServiceDeviceMode::RecoverTransport(vector<LinkData> &links, vector<std::pair<LinkGroup, u32>> linkGroupPair)
233 : {
234 3 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
235 :
236 : // ccu支持快照保存和恢复
237 2 : RecoverCcuTransport(links, linkGroupPair);
238 :
239 0 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
240 0 : }
241 :
242 1 : void CollServiceDeviceMode::RecoverCcuTransport(vector<LinkData> &links, vector<std::pair<LinkGroup, u32>> linkGroupPair)
243 : {
244 3 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
245 :
246 : // 创建TransPort所需的Socket
247 1 : comm->GetSocketManager().BatchCreateSockets(links);
248 :
249 1 : auto ret = GetCcuInsPreprocessor()->RecoverCcuTransportCtx(links, linkGroupPair);
250 1 : if (ret != HcclResult::HCCL_SUCCESS) {
251 1 : auto msg = StringFormat("Error occurs when call CollServiceDeviceMode::%s, error code: %d", __func__, ret);
252 1 : THROW<InternalException>(msg);
253 1 : }
254 :
255 0 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
256 0 : }
257 :
258 : // 功能说明：等待transport建链完成
259 : // 输入说明：string &opTag：通信域ID，唯一标记一个通信域
260 2 : bool CollServiceDeviceMode::IsAllTransportRecoveredReady(const std::string &opTag)
261 : {
262 2 : auto ret = GetCcuInsPreprocessor()->RecoverCcuTransportConfirm();
263 2 : if (ret == HcclResult::HCCL_SUCCESS) {
264 3 : HCCL_INFO("[CollServiceDeviceMode][IsAllTransportRecoveredReady] opTag[%s] recover transport success", opTag.c_str());
265 1 : return true;
266 : }
267 :
268 3 : HCCL_ERROR("[CollServiceDeviceMode][IsAllTransportRecoveredReady] fail, ret[%d]", ret);
269 1 : return false;
270 : }
271 :
272 :
273 0 : void CollServiceDeviceMode::RecoverAicpuTransport(vector<LinkData> &links) const
274 : {
275 0 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
276 :
277 : // 创建TransPort所需的Socket
278 0 : comm->GetSocketManager().BatchCreateSockets(links);
279 :
280 : // 创建TransPort所需的RmaConnection
281 0 : ConnectionsBuilder connectionsBuilder(*comm);
282 0 : connectionsBuilder.BatchBuild(comm->GetId(), links);
283 :
284 : // 创建TransPort所需的Notify资源
285 0 : RecoverInterRankNotifies(links);
286 :
287 : // 重新构造TransPort
288 0 : auto op = comm->GetCurrentCollOperator();
289 0 : if (op->opMode == OpMode::OPBASE) {
290 0 : comm->GetMemTransportManager()->BatchRecoverOpbasedTransports(links);
291 0 : } else if (op->opMode == OpMode::OFFLOAD) {
292 0 : comm->GetMemTransportManager()->BatchRecoverOffloadTransports(op->opTag, links);
293 : }
294 :
295 0 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
296 0 : }
297 :
298 0 : void CollServiceDeviceMode::RecoverInterRankNotifies(const vector<LinkData> &links) const
299 : {
300 0 : for (auto &link : links) {
301 : // 待修改: 申请数量
302 0 : comm->GetConnLocalNotifyManager().ApplyFor(link.GetRemoteRankId(), link);
303 : }
304 0 : }
305 :
306 : constexpr u32 TEMP_UES_CNTCKE_NUM = 16;
307 :
308 2 : HcclResult CollServiceDeviceMode::GetSnapShotDynamicBuf(CollOperator &op, BinaryStream &buf)
309 : {
310 6 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
311 :
312 2 : auto req = comm->GetCollAlgComponent()->GetCollAlgOpReq(op, comm->GetCurAlgName());
313 6 : HCCL_INFO("CollServiceAiCpuImpl::GetSnapShotDynamicBuf GetCollAlgOpReq OrchestMode::INSTRUCTION, algName %s",
314 : req.algName.c_str());
315 2 : buf << req.resReq.levelRankPairs.size();
316 3 : for (auto levelRankPair : req.resReq.levelRankPairs) {
317 1 : buf << levelRankPair.first << levelRankPair.second;
318 : }
319 :
320 2 : auto transportLinkGroup = ccuInsPreprocessor.GetCcuComm()->GetCcuTransportGrpMgr()->GetAllTransportGroups();
321 2 : vector<std::pair<LinkGroup, u32>> linkGroupPairs;
322 :
323 : // 临时规避多轮不同算子导致CNTCKE资源不足，cntCkeNum采用硬编码形式，待后续正式方案修改
324 4 : for (LinkGroup &group : transportLinkGroup) {
325 2 : linkGroupPairs.push_back({group, TEMP_UES_CNTCKE_NUM});
326 : }
327 2 : buf << linkGroupPairs.size();
328 6 : HCCL_INFO("[CollServiceDeviceMode::%s] linkGroupPairs size[%zu].", __func__, linkGroupPairs.size());
329 4 : for (auto linkGroupPair : linkGroupPairs) {
330 2 : LinkGroup &linkGroup = linkGroupPair.first;
331 2 : u32 cntCkeNum = linkGroupPair.second;
332 2 : buf << linkGroup.GetLinks().size();
333 6 : HCCL_INFO("[CollServiceDeviceMode::%s] linkGroup size[%zu].", __func__, linkGroup.GetLinks().size());
334 4 : for (auto &linkInfo : linkGroup.GetLinks()) {
335 2 : buf << linkInfo.rankId << linkInfo.dieId;
336 2 : linkInfo.localAddr.GetBinStream(buf);
337 2 : linkInfo.remoteAddr.GetBinStream(buf);
338 6 : HCCL_INFO("[CollServiceDeviceMode::%s] rankId[%d], dieId[%u], localAddr[%s], remoteAddr[%s].",
339 : __func__, linkInfo.rankId, linkInfo.dieId, linkInfo.localAddr.Describe().c_str(),
340 : linkInfo.remoteAddr.Describe().c_str());
341 2 : }
342 2 : buf << cntCkeNum;
343 2 : }
344 :
345 6 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
346 2 : return HcclResult::HCCL_SUCCESS;
347 2 : }
348 :
349 4 : void CollServiceDeviceMode::AllocCommResource(void *mc2Tiling, void **commContext, const AcceleratorState& tilingAccelerator)
350 : {
351 12 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
352 4 : if (tilingAccelerator == AcceleratorState::AIV || tilingAccelerator == AcceleratorState::AIV_ONLY) {
353 3 : aivMc2Compont.AllocCommResource(mc2Tiling, commContext);
354 : } else {
355 1 : mc2Compont.AllocCommResource(mc2Tiling, commContext);
356 : }
357 :
358 9 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
359 3 : }
360 :
361 1 : void CollServiceDeviceMode::GetCcuTaskInfo(void *tilingData, void *ccuTaskGroup)
362 : {
363 3 : HCCL_INFO("[CollServiceDeviceMode::%s] start.", __func__);
364 :
365 1 : auto taskParams = mc2Compont.GetCcuTaskInfo(tilingData);
366 1 : if (taskParams.size() > FUSION_SUB_TASK_MAX_CCU_NUM) {
367 0 : THROW<InternalException>(StringFormat("Get %d task params, which is bigger than the maximum size %d.",
368 : taskParams.size(), FUSION_SUB_TASK_MAX_CCU_NUM));
369 : }
370 :
371 1 : auto group = static_cast<rtCcuTaskGroup_t *>(ccuTaskGroup);
372 1 : group->taskNum = taskParams.size();
373 :
374 1 : for (size_t index = 0; index < taskParams.size(); ++index) {
375 0 : group->ccuTaskInfo[index].dieId = taskParams[index].dieId;
376 0 : group->ccuTaskInfo[index].missionId = taskParams[index].missionId;
377 0 : group->ccuTaskInfo[index].timeout = taskParams[index].timeout;
378 0 : group->ccuTaskInfo[index].instStartId = taskParams[index].instStartId;
379 0 : group->ccuTaskInfo[index].instCnt = taskParams[index].instCnt;
380 0 : group->ccuTaskInfo[index].key = taskParams[index].key;
381 0 : group->ccuTaskInfo[index].argSize = taskParams[index].argSize;
382 0 : std::copy(std::begin(taskParams[index].args), std::end(taskParams[index].args),
383 0 : std::begin(group->ccuTaskInfo[index].args));
384 0 : HCCL_INFO("ccu task info, dieId[%u] missionId[%u] instStartId[%u] instCnt[%u]", taskParams[index].dieId,
385 : taskParams[index].missionId, taskParams[index].instStartId, taskParams[index].instCnt);
386 : }
387 :
388 3 : HCCL_INFO("[CollServiceDeviceMode::%s] end.", __func__);
389 1 : }
390 :
391 4 : u32 CollServiceDeviceMode::GetCcuMc2ServerNum()
392 : {
393 4 : return mc2Compont.GetCcuMc2ServerNum();
394 : }
395 :
396 47 : CcuInsPreprocessor *CollServiceDeviceMode::GetCcuInsPreprocessor()
397 : {
398 47 : return &ccuInsPreprocessor;
399 : }
400 :
401 10 : AivInsPreprocessor *CollServiceDeviceMode::GetAivInsPreprocessor()
402 : {
403 10 : return &aivInsPreprocessor;
404 : }
405 :
406 1 : AicpuInsPreprocessor *CollServiceDeviceMode::GetAicpuInsPreprocessor()
407 : {
408 1 : return &aicpuInsPreprocessor;
409 : }
410 :
411 1 : bool CollServiceDeviceMode::IsAicpuResExisted(std::string algName)
412 : {
413 1 : return aicpuInsPreprocessor.IsAicpuResExisted(algName);
414 : }
415 :
416 1 : DevBuffer *CollServiceDeviceMode::GetAicpuResBuffer(std::string algName)
417 : {
418 1 : return aicpuInsPreprocessor.GetAicpuResBuffer(algName);
419 : }
420 :
421 : constexpr u32 TEMP_MAX_CNTCKE_NUM = 16; // 临时规避多轮不同算子导致CNTCKE资源不足，待后续正式方案修改
422 :
423 2 : void CollServiceDeviceMode::Resume()
424 : {
425 2 : CcuCommunicator *ccuComm = ccuInsPreprocessor.GetCcuComm();
426 2 : CHECK_NULLPTR(ccuComm, "[CollServiceDeviceMode::Resume] ccuComm is nullptr!");
427 :
428 2 : CcuTransportMgr *ccuTransportMgr = ccuComm->GetCcuTransportMgr();
429 2 : CHECK_NULLPTR(ccuTransportMgr, "[CollServiceDeviceMode::Resume] ccuTransportMgr is nullptr!");
430 2 : ccuTransportMgr->Resume();
431 2 : ccuTransportMgr->Confirm();
432 6 : HCCL_INFO("[CollServiceDeviceMode][%s] resource confirm end.", __func__);
433 :
434 2 : int32_t devLogicId = HrtGetDevice();
435 6 : for (uint8_t dieId = 0; dieId < MAX_CCU_IODIE_NUM; ++dieId) {
436 4 : CHK_RET_THROW(InternalException,
437 : StringFormat("[CollServiceDeviceMode][%s]Error occurs when call CcuCleanDieCkes, "
438 : "die[%u], devLogicId[%d].", __func__, dieId, devLogicId),
439 : CcuCleanDieCkes(devLogicId, dieId));
440 : }
441 2 : }
442 :
443 1 : HcclResult CollServiceDeviceMode::HandleAclGraphFirstOpAivBuff(rtStream_t mainStream)
444 : {
445 1 : rtModel_t rtModel = nullptr;
446 1 : bool isCapture = false;
447 1 : u32 modelId = 0;
448 1 : CHK_RET(GetStreamCaptureInfo(mainStream, rtModel, isCapture));
449 1 : if (isCapture) {
450 1 : CHK_PTR_NULL(rtModel);
451 : // 获取不到modelId会报错
452 1 : CHK_RET(GetModelId(rtModel, modelId));
453 1 : if (captureModelIds.find(modelId) == captureModelIds.end()) {
454 : // aclgraph场景，首算子清理AIV buff
455 1 : comm->SetAivClearEnable(true);
456 1 : comm->SetAivTag(1);
457 1 : captureModelIds.insert(modelId);
458 3 : HCCL_INFO("[CollServiceDeviceMode][%s] modelId[%u] is inserted to captureModelIds_", __func__, modelId);
459 : }
460 : }
461 1 : return HCCL_SUCCESS;
462 : }
463 :
464 1 : HcclResult CollServiceDeviceMode::GenerateAivOpArgs(const AivInstruction &aivInstruction, AivOpArgs &aivOpArgs) const
465 : {
466 1 : aivInstruction.GetAivInsArgs(aivOpArgs);
467 1 : aivOpArgs.aivTag = aivOpArgs.isOpBase ? (static_cast<uint32_t>(comm->GetAivTag()) << AIV_TAG_MOVE_LEFT_BITS) | static_cast<uint32_t>(aivOpArgs.aivTag):
468 1 : (static_cast<uint32_t>(comm->GetAivOffloadTag()) << AIV_TAG_MOVE_LEFT_BITS) | static_cast<uint32_t>(aivOpArgs.aivTag);
469 : void *buffersInAddr;
470 : void *buffersInAddrSrc;
471 1 : u64 buffersIn[MAX_RANK_SIZE_] = {};
472 1 : if (static_cast<unsigned int>(comm->GetMyRank()) >= MAX_RANK_SIZE_) {
473 0 : HCCL_ERROR("[CollServiceDeviceMode][%s] myRank is greater than or equal MAX_RANK_SIZE", __func__);
474 0 : return HCCL_E_PARA;
475 : }
476 1 : buffersIn[comm->GetMyRank()] = comm->GetCclBuffer()->GetAddr();
477 1 : auto ubMemLink2TransportMap = comm->GetUbMemoryTransportMgr()->GetRmtRankId2RmtIpcRmaBufList();
478 1 : for (auto ubMemLink2TransportIter : ubMemLink2TransportMap) {
479 0 : auto rmtRank = ubMemLink2TransportIter.first;
480 0 : auto rmtMemBuffer = ubMemLink2TransportIter.second->GetAddr();
481 0 : if (static_cast<unsigned int>(rmtRank) >= MAX_RANK_SIZE_) {
482 0 : HCCL_ERROR("[CollServiceDeviceMode][%s] rmtRank is greater than or equal MAX_RANK_SIZE", __func__);
483 0 : return HCCL_E_PARA;
484 : }
485 0 : buffersIn[rmtRank] = rmtMemBuffer;
486 : }
487 :
488 1 : buffersInAddr = aivOpArgs.isOpBase ? reinterpret_cast<void*>(comm->GetAivTagBuffer()->GetAddr()) : reinterpret_cast<void*>(comm->GetAivOffloadTagBuffer()->GetAddr());
489 3 : HCCL_INFO("%s AivTag[%u]", __func__, aivOpArgs.aivTag);
490 1 : aivOpArgs.buffersIn = buffersInAddr;
491 1 : HrtMemcpy(buffersInAddr, MAX_RANK_SIZE_ * sizeof(uint64_t), buffersIn, MAX_RANK_SIZE_ * sizeof(uint64_t),
492 : RT_MEMCPY_HOST_TO_DEVICE);
493 1 : u64 buffersOut[MAX_RANK_SIZE_] = {};
494 1 : auto ubMemLink2TransportMap_ = aivOpArgs.isOpBase ? comm->GetUbMemoryTransportMgr()->GetAllRankId2AivTagBufAddrList():
495 1 : comm->GetUbMemoryTransportMgr()->GetAllRankId2AivOffloadTagBufAddrList();
496 2 : for (auto ubMemLink2TransportIter : ubMemLink2TransportMap_) {
497 1 : auto rmtRank = ubMemLink2TransportIter.first;
498 1 : auto rmtMemBuffer = ubMemLink2TransportIter.second;
499 1 : if (static_cast<unsigned int>(rmtRank) >= MAX_RANK_SIZE_) {
500 0 : HCCL_ERROR("[CollServiceDeviceMode][%s] rmtRank is greater than or equal MAX_RANK_SIZE", __func__);
501 0 : return HCCL_E_PARA;
502 : }
503 1 : buffersOut[rmtRank] = rmtMemBuffer;
504 : }
505 1 : buffersInAddr = aivOpArgs.isOpBase ? reinterpret_cast<void*>(comm->GetAivTagBuffer()->GetAddr() + AIV_TAG_ADDR_OFFSET) :
506 1 : reinterpret_cast<void*>(comm->GetAivOffloadTagBuffer()->GetAddr() + AIV_TAG_ADDR_OFFSET);
507 1 : HrtMemcpy(buffersInAddr, MAX_RANK_SIZE_ * sizeof(uint64_t), buffersOut, MAX_RANK_SIZE_ * sizeof(uint64_t),
508 : RT_MEMCPY_HOST_TO_DEVICE);
509 :
510 1 : buffersInAddr = aivOpArgs.isOpBase ? reinterpret_cast<void *>(comm->GetAivTagBuffer()->GetAddr() + AIV_FLAG_ADDR_OFFSET):
511 1 : reinterpret_cast<void *>(comm->GetAivOffloadTagBuffer()->GetAddr() + AIV_FLAG_ADDR_OFFSET);
512 1 : buffersInAddrSrc = aivOpArgs.isOpBase ? reinterpret_cast<void *>(comm->GetAivTagBuffer()->GetAddr() + AIV_FLAG_CLEAR_OFFSET):
513 1 : reinterpret_cast<void *>(comm->GetAivOffloadTagBuffer()->GetAddr() + AIV_FLAG_CLEAR_OFFSET);
514 1 : bool isAivClearEnable = comm->GetAivClearEnable();
515 1 : if (isAivClearEnable && (aivOpArgs.aivTag & AIV_LOW_16_BITS) == 1 && (aivOpArgs.aivTag >> AIV_TAG_MOVE_LEFT_BITS) == 1) {
516 0 : HrtMemcpy(buffersInAddr, AIV_FLAG_AREA_SIZE, buffersInAddrSrc, AIV_FLAG_AREA_SIZE, RT_MEMCPY_DEVICE_TO_DEVICE);
517 : }
518 1 : if (comm->GetCurrentCollOperator()->inputMem == nullptr) {
519 0 : HCCL_INFO("%s comm->GetCurrentCollOperator()->inputMem is nullptr", __func__);
520 : } else {
521 1 : u64 localInputAddr = static_cast<uint64_t>(comm->GetCurrentCollOperator()->inputMem->GetAddr());
522 1 : aivOpArgs.input += localInputAddr;
523 : }
524 :
525 1 : if (comm->GetCurrentCollOperator()->outputMem == nullptr) {
526 0 : HCCL_INFO("%s comm->GetCurrentCollOperator()->outputMem is nullptr", __func__);
527 : } else {
528 1 : u64 localOutputAddr = static_cast<uint64_t>(comm->GetCurrentCollOperator()->outputMem->GetAddr());
529 1 : aivOpArgs.output += localOutputAddr;
530 : }
531 1 : return HCCL_SUCCESS;
532 1 : }
533 :
534 1 : void CollServiceDeviceMode::GeneratorAivSuperKernelArgs(const AivOpArgs &aivOpArgs, bool clearEnable, u32 numBlocks,
535 : AivSuperKernelArgs &superArgs) const
536 : {
537 1 : auto op = comm->GetCurrentCollOperator();
538 1 : superArgs.buffersIn = aivOpArgs.buffersIn;
539 1 : superArgs.rank = comm->GetMyRank();
540 1 : superArgs.rankSize = comm->GetRankSize();
541 1 : u64 dataCount = 0;
542 1 : DataType dataType = Hccl::DataType::INVALID;
543 1 : if (op->opType == OpType::ALLTOALL) {
544 0 : dataCount = op->all2AllDataDes.sendCount;
545 0 : dataType = op->all2AllDataDes.sendType;
546 : } else {
547 1 : dataCount = op->dataCount;
548 1 : dataType = op->dataType;
549 : }
550 1 : superArgs.len = dataCount;
551 1 : superArgs.dataType = dataType;
552 1 : superArgs.unitSize = SIZE_TABLE_ORION[dataType];
553 1 : superArgs.reduceOp = op->reduceOp;
554 1 : superArgs.numBlocks = numBlocks;
555 1 : superArgs.tag = comm->GetAivTag();
556 1 : superArgs.clearEnable = (clearEnable ? 1 : 0);
557 1 : superArgs.inputSliceStride = 0;
558 1 : superArgs.outputSliceStride = 0;
559 1 : superArgs.repeatNum = 1;
560 1 : superArgs.inputRepeatStride = 0;
561 1 : superArgs.outputRepeatStride = 0;
562 1 : superArgs.input = aivOpArgs.input;
563 1 : superArgs.output = aivOpArgs.output;
564 1 : superArgs.cclBufferSize = comm->GetBufferSize();
565 :
566 3 : HCCL_INFO("[CollServiceDeviceMode::%s] Tag %lld, clearEnable %lld, numBlocks %llu, dataCount %llu, cclBufferSize %llu.", __func__, superArgs.tag,
567 : superArgs.clearEnable, superArgs.numBlocks, dataCount, superArgs.cclBufferSize);
568 1 : }
569 :
570 1 : HcclResult CollServiceDeviceMode::GetAlgExecParam(bool clearEnable, u32 numBlocks, void *&commContext, u64 &len)
571 : {
572 1 : auto op = comm->GetCurrentCollOperator();
573 3 : HCCL_INFO("[CollServiceDeviceMode][%s] op[%p] sendCount[%u], recvCount[%u]",
574 : __func__, op,
575 : op->all2AllDataDes.sendCount,
576 : op->all2AllDataDes.recvCount);
577 : // 建链
578 1 : shared_ptr<InsQueue> insQueue = Orchestrate(*op);
579 1 : AllocQueueNotify(*insQueue);
580 1 : std::vector<LinkData> uniqueLinks = GetUniqueLinks(insQueue);
581 : // Socket建链
582 1 : comm->GetSocketManager().BatchCreateSockets(uniqueLinks);
583 1 : GetAivInsPreprocessor()->Preprocess(insQueue);
584 : // 组装AivOpArgs
585 1 : AivOpArgs aivOpArgs{};
586 1 : for (auto ins = insQueue->Iter(); ins.HasNext(); ++ins) {
587 1 : if (ins->GetType() != InstructionType::AIV_INS) {
588 0 : continue;
589 : }
590 1 : const AivInstruction &aivIns = dynamic_cast<const AivInstruction &>(*ins);
591 1 : CHK_RET(GenerateAivOpArgs(aivIns, aivOpArgs));
592 1 : break;
593 1 : }
594 :
595 : // aivOpArgs转为aivSuperKernelArgs的参数
596 1 : AivSuperKernelArgs aivSuperKernelArgs{};
597 1 : GeneratorAivSuperKernelArgs(aivOpArgs, clearEnable, numBlocks, aivSuperKernelArgs);
598 :
599 1 : void *sendAlgParamMemPtr = nullptr;
600 : // alloc device 地址
601 1 : sendAlgParamMemPtr = HrtMalloc(sizeof(AivSuperKernelArgs), static_cast<int>(ACL_MEM_TYPE_HIGH_BAND_WIDTH));
602 1 : CHK_PTR_NULL(sendAlgParamMemPtr);
603 3 : HCCL_INFO("SPK sendalgparam %p.", sendAlgParamMemPtr);
604 :
605 : // 拷贝到Device
606 1 : HrtMemcpy(sendAlgParamMemPtr, sizeof(AivSuperKernelArgs), &aivSuperKernelArgs, sizeof(AivSuperKernelArgs),
607 : RT_MEMCPY_HOST_TO_DEVICE);
608 1 : commContext = sendAlgParamMemPtr;
609 1 : len = sizeof(AivSuperKernelArgs);
610 1 : return HCCL_SUCCESS;
611 1 : }
612 :
613 : } // namespace Hccl
|
