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