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 "communicator_impl_lite.h"
12 : #include "aicpu_res_package_helper.h"
13 : #include "alg_topo_package_helper.h"
14 : #include "sal.h"
15 : #include "suspending_exception.h"
16 : #include "exception_util.h"
17 : #include "task_info.h"
18 : namespace Hccl {
19 :
20 : constexpr int KERNEL_SUCCESS_CODE = 0;
21 : constexpr int KERNEL_ERROR_CODE = 1;
22 : constexpr int ALLTOALLV_DATA_INDEX_2 = 2; // sdispls在sendRecvInfos数组中的偏移
23 : constexpr int ALLTOALLV_DATA_INDEX_3 = 3; // rdispls在sendRecvInfos数组中的偏移
24 :
25 5 : HcclResult CommunicatorImplLite::InitProfilingReporterLite()
26 : {
27 20 : CHK_RET(Hccl::ProfilingHandlerLite::GetInstance().Init());
28 0 : CHK_RET(profilingReporterLite->Init());
29 0 : return HCCL_SUCCESS;
30 : }
31 :
32 5 : int CommunicatorImplLite::LoadWithOpBasedMode(HcclKernelParamLite *kernelParam)
33 : {
34 : try {
35 : // 设定devType，初始化能力，算法及其他模块通过Get获取能力
36 5 : DevCapability::GetInstance().Init(kernelParam->comm.devType);
37 5 : HcclResult ret = InitProfilingReporterLite();
38 5 : if (ret != HCCL_SUCCESS) {
39 15 : HCCL_ERROR("InitProfilingReporterLite failed, ret[%d]", ret);
40 5 : return KERNEL_ERROR_CODE;
41 : }
42 0 : UnfoldOp(kernelParam);
43 0 : } catch (HcclException &e) {
44 0 : HCCL_ERROR("Hccl exception %s was caught.", e.what());
45 0 : return KERNEL_ERROR_CODE;
46 0 : } catch (std::exception &e) {
47 0 : HCCL_ERROR("Std exception %s was caught.", e.what());
48 0 : return KERNEL_ERROR_CODE;
49 0 : } catch (...) {
50 0 : HCCL_ERROR("Some unknown error ocured.");
51 0 : return KERNEL_ERROR_CODE;
52 0 : }
53 :
54 0 : return KERNEL_SUCCESS_CODE;
55 : }
56 :
57 5 : int CommunicatorImplLite::UpdateComm(HcclKernelParamLite *kernelParam)
58 : {
59 5 : if(!isSuspended){
60 6 : HCCL_ERROR("CommunicatorImplLite is not suspended");
61 2 : return KERNEL_ERROR_CODE;
62 : }
63 : try {
64 : // 设定devType，初始化能力，算法及其他模块通过Get获取能力
65 3 : DevCapability::GetInstance().Init(kernelParam->comm.devType);
66 3 : UpdateTransports(kernelParam);
67 :
68 1 : auto id = GetHostDeviceSyncNotifyLiteMgr()->GetDeviceWaitNotify()->GetId();
69 1 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[UpdateComm] master stream is nullptr!");
70 1 : streamLiteMgr->GetMaster()->GetRtsq()->NotifyWait(id);
71 1 : id = GetHostDeviceSyncNotifyLiteMgr()->GetHostWaitNotify()->GetId();
72 1 : streamLiteMgr->GetMaster()->GetRtsq()->NotifyRecordLoc(id);
73 1 : streamLiteMgr->GetMaster()->GetRtsq()->LaunchTask();
74 3 : HCCL_INFO("[NsRecovery] UpdateComm: task launched.");
75 2 : } catch (HcclException &e) {
76 0 : HCCL_ERROR("CommunicatorImplLite::UpdateComm Hccl exception %s was caught.", e.what());
77 0 : return KERNEL_ERROR_CODE;
78 1 : } catch (std::exception &e) {
79 3 : HCCL_ERROR("CommunicatorImplLite::UpdateComm Std exception %s was caught.", e.what());
80 1 : return KERNEL_ERROR_CODE;
81 2 : } catch (...) {
82 3 : HCCL_ERROR("CommunicatorImplLite::UpdateComm Some unknown error ocured.");
83 1 : return KERNEL_ERROR_CODE;
84 1 : }
85 1 : isSuspended = false;
86 1 : return KERNEL_SUCCESS_CODE;
87 : }
88 :
89 0 : std::shared_ptr<InsQueue> CommunicatorImplLite::GetInsQueue(HcclKernelParamLite *kernelParam)
90 : {
91 0 : if (kernelParam->oneSidedComm) {
92 0 : HCCL_INFO("CommunicatorImplLite::GetInsQueue oneSidedComm begin");
93 0 : CreateOneSidedComponentLite();
94 0 : return GetOneSidedInsQueue(kernelParam);
95 : }
96 :
97 0 : CreateCollAlgComponentLite();
98 0 : HCCL_INFO("CommunicatorImplLite::GetInsQueue begin kernelParam->algName = %s", kernelParam->algName);
99 0 : std::shared_ptr<InsQueue> queue = std::make_shared<InsQueue>();
100 0 : auto it = algTopoInfoMap.find(kernelParam->tagKey);
101 0 : auto ret = algComponentLite->Orchestrate(kernelParam->op.algOperator, kernelParam->algName, it->second, queue);
102 0 : if (ret == HCCL_E_PARA) {
103 0 : return nullptr;
104 : }
105 0 : return queue;
106 0 : }
107 :
108 0 : void CommunicatorImplLite::CreateCollAlgComponentLite()
109 : {
110 0 : if (algComponentLite.get() == nullptr) {
111 : algComponentLite
112 0 : = make_unique<CollAlgComponentLite>(myRank, rankSize, devType, scratchSize, connectedLinkMgr.get(),
113 0 : rmtDataBufferMgr.get());
114 0 : HCCL_INFO("CommunicatorImplLite::CreateCollAlgComponentLite is null");
115 : } else {
116 0 : algComponentLite->UpdateScratchBufferSize(scratchSize);
117 0 : HCCL_INFO("CommunicatorImplLite::CreateCollAlgComponentLite, bufferSize %llu", scratchSize);
118 : }
119 0 : }
120 :
121 0 : void CommunicatorImplLite::UnfoldOp(HcclKernelParamLite *kernelParam)
122 : {
123 0 : opIndex_ = kernelParam->comm.opIndex_;
124 0 : uint64_t beginTime = ProfGetCurCpuTimestamp();
125 0 : profilingReporterLite->UpdateProfStat();
126 0 : UpdateCommParam(kernelParam);
127 0 : UpdateLocBuffer(kernelParam);
128 0 : UpdateUserStreamId(kernelParam);
129 0 : UpdateRes(kernelParam);
130 0 : UpdateDynamicOpData(kernelParam);
131 0 : SetDfxOpInfo(beginTime);
132 :
133 0 : UpdateHDCommnicate(kernelParam);
134 0 : RegisterRtsqCallback();
135 :
136 0 : isCommReady = true;
137 0 : HCCL_INFO("CommunicatorImplLite::UnfoldOpBase isCommReady is set to true.");
138 0 : std::shared_ptr<InsQueue> insQueue = GetInsQueue(kernelParam);
139 0 : if (insQueue == nullptr) {
140 0 : THROW<NullPtrException>(StringFormat("CommunicatorImplLite::UnfoldOpBase insQueue is nullptr."));
141 : }
142 0 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960) {
143 0 : HCCL_INFO("CommunicatorImplLite::UnfoldOpBase DevType is DEV_TYPE_950 or DEV_TYPE_960.");
144 0 : insExecutor->ExecuteV82(*insQueue);
145 0 : profilingReporterLite->ReportAllTasks();
146 0 : ProfilingHandlerLite::GetInstance().ReportHcclOpInfo(*mirrorTaskMgrLite->GetCurrDfxOpInfo());
147 0 : } else if (devType == DevType::DEV_TYPE_910A2) {
148 0 : HCCL_INFO("CommunicatorImplLite::UnfoldOpBase DevType is DEV_TYPE_910A2.");
149 0 : insExecutor->Execute(*insQueue);
150 : } else {
151 0 : HCCL_WARNING("CommunicatorImplLite::UnfoldOpBase DevType is not support.");
152 : }
153 0 : kernelParam->op.algOperator.scratchMem = nullptr;
154 0 : }
155 :
156 0 : void CommunicatorImplLite::RegisterRtsqCallback()
157 : {
158 0 : auto checkOpExecStatusCallback = [this](){ this->CheckOpExecStatus(); };
159 0 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[RegisterRtsqCallback]master stream is nullptr!");
160 0 : streamLiteMgr->GetMaster()->GetRtsq()->SetOpExecStatusCallback(checkOpExecStatusCallback);
161 0 : for (u32 i = 0; i < streamLiteMgr->SizeOfSlaves(); ++i) {
162 0 : streamLiteMgr->GetSlave(i)->GetRtsq()->SetOpExecStatusCallback(checkOpExecStatusCallback);
163 : }
164 0 : }
165 :
166 0 : void CommunicatorImplLite::CheckOpExecStatus() const
167 : {
168 0 : if (isSuspended) {
169 0 : HCCL_INFO("hccl aicpu stop wait finish, for recv stop launch cmd");
170 0 : THROW<SuspendingException>(StringFormat("[CheckOpExecStatus] recv stop launch command, coll service is suspended."));
171 : }
172 0 : }
173 :
174 5 : void CommunicatorImplLite::UpdateCommParam(HcclKernelParamLite *kernelParam)
175 : {
176 5 : if (isUpdateComm) {
177 0 : return;
178 : }
179 5 : myRank = kernelParam->comm.myRank;
180 5 : rankSize = kernelParam->comm.rankSize;
181 5 : devPhyId = kernelParam->comm.devPhyId;
182 5 : devType = kernelParam->comm.devType;
183 5 : opCounterAddr = kernelParam->comm.opCounterAddr;
184 5 : hcclExecTimeout = kernelParam->envConfig.hcclExecTimeout;
185 5 : if (rmtDataBufferMgr == nullptr) {
186 4 : collAlgInfo = std::make_unique<CollAlgInfo>(kernelParam->op.algOperator.opMode, kernelParam->opTag);
187 4 : rmtDataBufferMgr = std::make_unique<RmtDataBufferMgr>(transportLiteMgr.get(), collAlgInfo.get());
188 : }
189 5 : commId = kernelParam->comm.commId;
190 15 : HCCL_INFO(
191 : "CommunicatorImplLite::UpdateCommParam myRank [%u] rankSize[%u] devPhyId[%u] devType[%d] scratchSize [%llu] "
192 : "scratchaddress[%llx] opCounterAddr[%llx] commId[%s]",
193 : myRank, rankSize, devPhyId, devType, scratchSize, kernelParam->comm.opBaseScratch.addr, opCounterAddr,
194 : commId.c_str());
195 5 : isUpdateComm = true;
196 : }
197 :
198 3 : void CommunicatorImplLite::UpdateLocBuffer(HcclKernelParamLite *kernelParam)
199 : {
200 3 : locBuffer[BufferType::INPUT] = kernelParam->op.input.addr;
201 3 : locBuffer[BufferType::OUTPUT] = kernelParam->op.output.addr;
202 :
203 3 : rmaBufferLiteVec.clear();
204 3 : rmaBufferLiteVec.resize(BufferType::__COUNT__);
205 :
206 3 : InitRmaBufferLite(kernelParam->op.input, BufferType::INPUT);
207 3 : InitRmaBufferLite(kernelParam->op.output, BufferType::OUTPUT);
208 :
209 3 : if (kernelParam->op.algOperator.opMode == OpMode::OPBASE) {
210 0 : scratchSize = kernelParam->comm.opBaseScratch.size;
211 0 : locBuffer[BufferType::SCRATCH] = kernelParam->comm.opBaseScratch.addr;
212 0 : InitRmaBufferLite(kernelParam->comm.opBaseScratch, BufferType::SCRATCH);
213 : } else {
214 3 : scratchSize = kernelParam->op.scratch.size;
215 3 : locBuffer[BufferType::SCRATCH] = kernelParam->op.scratch.addr;
216 3 : InitRmaBufferLite(kernelParam->op.scratch, BufferType::SCRATCH);
217 : }
218 :
219 3 : if (kernelParam->oneSidedComm) {
220 0 : if ((kernelParam->op.batchPutGetLocalAddr == nullptr) || (kernelParam->op.batchPutGetRemoteAddr == nullptr)) {
221 0 : THROW<InternalException>("batchPutGetAddr is nullptr");
222 : }
223 : }
224 :
225 3 : InitCurrentOp(kernelParam);
226 9 : HCCL_INFO("CommunicatorImplLite::UpdateLocBuffer locBuffer[BufferType::INPUT] %llx, locBuffer[BufferType::OUTPUT] %llx",
227 : locBuffer[BufferType::INPUT], locBuffer[BufferType::OUTPUT]);
228 9 : HCCL_INFO("CommunicatorImplLite::UpdateLocBuffer locBuffer[BufferType::SCRATCH] %llx", locBuffer[BufferType::SCRATCH]);
229 3 : }
230 :
231 1 : void CommunicatorImplLite::UpdateTransports(HcclKernelParamLite *kernelParam)
232 : {
233 3 : HCCL_INFO("[NsRecovery] RestoreAllTransports start");
234 1 : RestoreAllTransports(kernelParam->binaryResAddr, kernelParam->binaryResSize);
235 3 : HCCL_INFO("[NsRecovery] RestoreAllTransports end");
236 1 : }
237 :
238 1 : void CommunicatorImplLite::RestoreAllTransports(u64 addr, u64 bufSize)
239 : {
240 1 : std::vector<char> data;
241 1 : data.resize(bufSize);
242 1 : int ret = memcpy_s(data.data(), bufSize, reinterpret_cast<void *>(addr), bufSize);
243 1 : if (ret != 0) {
244 0 : THROW<InternalException>(StringFormat("[NsRecovery] CommunicatorImplLite::RestoreAllTransports: memcpy_s failed, ret = %d", ret));
245 : }
246 3 : HCCL_INFO("[NsRecovery] CommunicatorImplLite::RestoreAllTransports: RestoreData %s", Bytes2hex(data.data(), data.size()).c_str());
247 : AicpuResPackageHelper helper;
248 1 : auto dataVec = helper.ParsePackedData(data);
249 :
250 1 : AicpuResMgrType resType = AicpuResMgrType::TRANSPORT;
251 1 : GetTransportLiteMgr()->ParseAllPackedData(dataVec[resType].data);
252 3 : HCCL_INFO("[NsRecovery] CommunicatorImplLite::RestoreAllTransports: GetResMgr %s Data", resType.Describe().c_str());
253 1 : }
254 :
255 0 : bool CommunicatorImplLite::CheckNeedUpdateRes(HcclKernelParamLite *kernelParam)
256 : {
257 0 : std::string tagKey = kernelParam->tagKey;
258 0 : auto it = loadedOpSet.find(tagKey);
259 0 : if (it != loadedOpSet.end()) {
260 0 : HCCL_INFO("[CheckNeedUpdateRes] Corresponding resources of tag[%s] have been loaded", tagKey.c_str());
261 0 : return false;
262 : }
263 0 : loadedOpSet.insert(tagKey);
264 0 : return true;
265 0 : }
266 :
267 0 : void CommunicatorImplLite::UpdateRes(HcclKernelParamLite *kernelParam)
268 : {
269 0 : if (CheckNeedUpdateRes(kernelParam)) {
270 0 : HCCL_INFO("[UpdateRes] start, opMode[%s]", kernelParam->op.algOperator.opMode.Describe().c_str());
271 0 : RestoreOpRes(kernelParam->opTag, kernelParam->tagKey, kernelParam->binaryResAddr, kernelParam->binaryResSize);
272 0 : HCCL_INFO("[UpdateRes] end");
273 : }
274 0 : }
275 :
276 0 : void CommunicatorImplLite::UpdateHDCommnicate(HcclKernelParamLite *kernelParam)
277 : {
278 0 : CHK_RET_THROW(InternalException, StringFormat("[CommunicatorImplLite][%s] failed to init kfcControlTransferH2DParams", __func__),
279 : kfcControlTransferH2D->Init(kernelParam->kfcControlTransferH2DParams));
280 0 : CHK_RET_THROW(InternalException, StringFormat("[CommunicatorImplLite][%s] failed to init kfcControlTransferD2HParams", __func__),
281 : kfcStatusTransferD2H->Init(kernelParam->kfcControlTransferD2HParams));
282 0 : std::unique_lock<std::mutex> lock(hdcShmLock_);
283 0 : hdcHandler = make_unique<AicpuHdcHandler>(*kfcControlTransferH2D, *kfcStatusTransferD2H);
284 0 : }
285 :
286 0 : void CommunicatorImplLite::UpdateDynamicOpData(HcclKernelParamLite *kernelParam) const
287 : {
288 0 : HCCL_INFO("[CommunicatorImplLite][UpdateDynamicOpData] OpType[%s]", kernelParam->op.algOperator.opType.Describe().c_str());
289 0 : u8* dynamicDataPtr = reinterpret_cast<u8*>(kernelParam) + sizeof(struct HcclKernelParamLite);
290 0 : if (kernelParam->op.algOperator.opType == OpType::BATCHSENDRECV) {
291 0 : struct BatchSendRecvDataDes* batchSendRecvDataPtr =
292 : reinterpret_cast<struct BatchSendRecvDataDes*>(dynamicDataPtr);
293 0 : kernelParam->op.algOperator.batchSendRecvDataDes.itemNum = batchSendRecvDataPtr->itemNum;
294 0 : kernelParam->op.algOperator.batchSendRecvDataDes.sendRecvItemsPtr = batchSendRecvDataPtr->batchSendRecvItem;
295 0 : HcclSendRecvItem* itemPtr = reinterpret_cast<HcclSendRecvItem *>(kernelParam->op.algOperator.batchSendRecvDataDes.sendRecvItemsPtr);
296 0 : for (u32 i = 0; i < kernelParam->op.algOperator.batchSendRecvDataDes.itemNum; i++) {
297 0 : HCCL_INFO("[CommunicatorImplLite][UpdateDynamicOpData] batchSendRecvDataDes remoteRank[%u]", (itemPtr + i)->remoteRank);
298 : }
299 0 : } else if (kernelParam->op.algOperator.opType == OpType::ALLTOALLV) {
300 0 : struct AllToAllvDataDes* alltoallvDataPtr =
301 : reinterpret_cast<struct AllToAllvDataDes *>(dynamicDataPtr);
302 0 : kernelParam->op.algOperator.all2AllVDataDes.sendType = static_cast<DataType::Value>(alltoallvDataPtr->sendType);
303 0 : kernelParam->op.algOperator.all2AllVDataDes.recvType = static_cast<DataType::Value>(alltoallvDataPtr->recvType);
304 0 : u64 rankSize = kernelParam->comm.rankSize;
305 0 : kernelParam->op.algOperator.all2AllVDataDes.sendCounts = static_cast<void *>(alltoallvDataPtr->sendRecvInfos);
306 0 : kernelParam->op.algOperator.all2AllVDataDes.recvCounts = static_cast<void *>(static_cast<u64 *>(alltoallvDataPtr->sendRecvInfos) + rankSize);
307 0 : kernelParam->op.algOperator.all2AllVDataDes.sdispls = static_cast<void *>(static_cast<u64 *>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_DATA_INDEX_2 * rankSize);
308 0 : kernelParam->op.algOperator.all2AllVDataDes.rdispls = static_cast<void *>(static_cast<u64 *>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_DATA_INDEX_3 * rankSize);
309 0 : for (u32 i = 0; i < rankSize; i++) {
310 0 : HCCL_INFO("[CommunicatorImplLite][UpdateDynamicOpData] alltoallv sendCounts[%llu], recvCounts[%llu]",
311 : *(static_cast<const u64 *>(kernelParam->op.algOperator.all2AllVDataDes.sendCounts) + i),
312 : *(static_cast<const u64 *>(kernelParam->op.algOperator.all2AllVDataDes.recvCounts) + i));
313 : }
314 0 : } else if (kernelParam->op.algOperator.opType == OpType::ALLTOALLVC) {
315 0 : struct AllToAllvcDataDes* alltoallvcDataPtr =
316 : reinterpret_cast<struct AllToAllvcDataDes *>(dynamicDataPtr);
317 0 : kernelParam->op.algOperator.all2AllVCDataDes.sendType = static_cast<DataType::Value>(alltoallvcDataPtr->sendType);
318 0 : kernelParam->op.algOperator.all2AllVCDataDes.recvType = static_cast<DataType::Value>(alltoallvcDataPtr->recvType);
319 0 : kernelParam->op.algOperator.all2AllVCDataDes.sendCountMatrix = static_cast<void *>(alltoallvcDataPtr->sendCountMatrix);
320 : }
321 0 : }
322 :
323 21 : HostDeviceSyncNotifyLiteMgr *CommunicatorImplLite::GetHostDeviceSyncNotifyLiteMgr()
324 : {
325 21 : return hostDeviceSyncNotifyLiteMgr.get();
326 : }
327 :
328 50 : StreamLiteMgr *CommunicatorImplLite::GetStreamLiteMgr()
329 : {
330 50 : return streamLiteMgr.get();
331 : }
332 :
333 1 : QueueNotifyLiteMgr *CommunicatorImplLite::GetQueueNotifyLiteMgr()
334 : {
335 1 : return queueNotifyLiteMgr.get();
336 : }
337 :
338 1 : Cnt1tonNotifyLiteMgr *CommunicatorImplLite::GetCnt1tonNotifyLiteMgr()
339 : {
340 1 : return cnt1tonNotifyLiteMgr.get();
341 : }
342 :
343 1 : CntNto1NotifyLiteMgr *CommunicatorImplLite::GetCntNto1NotifyLiteMgr()
344 : {
345 1 : return cntNto1NotifyLiteMgr.get();
346 : }
347 :
348 1 : ConnectedLinkMgr *CommunicatorImplLite::GetConnectedLinkMgr()
349 : {
350 1 : return connectedLinkMgr.get();
351 : }
352 :
353 89 : DevId CommunicatorImplLite::GetDevPhyId()
354 : {
355 89 : return devPhyId;
356 : }
357 :
358 0 : u32 CommunicatorImplLite::GetExecTimeOut()
359 : {
360 0 : return hcclExecTimeout;
361 : }
362 :
363 2 : KfcCommand CommunicatorImplLite::BackGroundGetCmd()
364 : {
365 2 : std::unique_lock<std::mutex> lock(hdcShmLock_);
366 4 : return hdcHandler->GetKfcCommand();
367 2 : }
368 :
369 2 : void CommunicatorImplLite::BackGroundSetStatus(KfcStatus status, KfcErrType errorCode)
370 : {
371 2 : std::unique_lock<std::mutex> lock(hdcShmLock_);
372 2 : hdcHandler->SetKfcExecStatus(status, errorCode);
373 2 : }
374 :
375 : // 从 buffer中解析出算子需要的信息 ，对应 Host侧的 PackOpData
376 0 : void CommunicatorImplLite::RestoreOpRes(const string &opTag, const string &tagKey, u64 addr, u64 bufSize)
377 : {
378 0 : std::vector<char> data;
379 0 : data.resize(bufSize);
380 0 : (void)memcpy_s(data.data(), bufSize, reinterpret_cast<void *>(addr), bufSize);
381 : AicpuResPackageHelper helper;
382 0 : auto dataVec = helper.ParsePackedData(data);
383 :
384 0 : AicpuResMgrType resType = AicpuResMgrType::ALG_COMP_INFO;
385 0 : CreateCollAlgComponentLite();
386 0 : if (dataVec[resType].data.size() != 0) {
387 0 : algComponentLite->ParsePackedData(dataVec[resType].data);
388 : }
389 :
390 0 : resType = AicpuResMgrType::STREAM;
391 0 : GetStreamLiteMgr()->ParsePackedData(dataVec[resType].data);
392 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
393 :
394 0 : resType = AicpuResMgrType::QUEUE_NOTIFY;
395 0 : GetQueueNotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
396 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
397 :
398 0 : resType = AicpuResMgrType::QUEUE_WAIT_GROUP_CNT_NOTIFY;
399 0 : GetCntNto1NotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
400 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
401 :
402 0 : resType = AicpuResMgrType::QUEUE_BCAST_POST_CNT_NOTIFY;
403 0 : GetCnt1tonNotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
404 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
405 :
406 0 : resType = AicpuResMgrType::HOST_DEV_SYNC_NOTIFY;
407 0 : GetHostDeviceSyncNotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
408 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
409 :
410 0 : resType = AicpuResMgrType::TRANSPORT;
411 0 : if (currentOp.opMode == OpMode::OPBASE) { // 单算子模式
412 0 : GetTransportLiteMgr()->ParseOpbasePackedData(dataVec[resType].data);
413 0 : } else if (currentOp.opMode == OpMode::OFFLOAD) { // 图下沉模式
414 0 : GetTransportLiteMgr()->ParseOffloadPackedData(opTag, dataVec[resType].data);
415 : } else {
416 0 : THROW<InternalException>(StringFormat("opMode=%s failed", currentOp.opMode.Describe().c_str()));
417 : }
418 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data, %s", opTag.c_str(), resType.Describe().c_str(),
419 : currentOp.opMode.Describe().c_str());
420 :
421 0 : resType = AicpuResMgrType::ALG_TOPO;
422 : AlgTopoPackageHelper algTopoHelper;
423 0 : algTopoInfoMap[tagKey] = algTopoHelper.GetAlgTopoInfo(dataVec[resType].data);
424 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data, tagKey=%s", opTag.c_str(), resType.Describe().c_str(),
425 : tagKey.c_str());
426 :
427 0 : resType = AicpuResMgrType::CONNECTD_MGR;
428 0 : GetConnectedLinkMgr()->ParsePackedData(dataVec[resType].data);
429 0 : HCCL_INFO("CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
430 0 : }
431 :
432 80 : CommunicatorImplLite::CommunicatorImplLite(u32 idIndex) : idIndex_(idIndex)
433 : {
434 80 : }
435 :
436 9 : void CommunicatorImplLite::InitRmaBufferLite(HcclAicpuLocBufLite &bufLite, BufferType type)
437 : {
438 9 : rmaBufferLiteVec[type]
439 18 : = std::make_unique<RmaBufferLite>(bufLite.addr, bufLite.size, bufLite.tokenId, bufLite.tokenValue);
440 9 : }
441 :
442 3 : void CommunicatorImplLite::InitCurrentOp(HcclKernelParamLite *kernelParam)
443 : {
444 3 : currentOp.opTag = kernelParam->opTag; // opTag的赋值
445 9 : HCCL_INFO("CommunicatorImplLite::InitCurrentOp opTag[%s]", currentOp.opTag.c_str());
446 :
447 3 : currentOp.opMode = kernelParam->op.algOperator.opMode;
448 3 : currentOp.opType = kernelParam->op.algOperator.opType;
449 3 : currentOp.reduceOp = kernelParam->op.algOperator.reduceOp;
450 3 : currentOp.dataType = kernelParam->op.algOperator.dataType;
451 3 : currentOp.outputDataType = kernelParam->op.algOperator.outputDataType;
452 3 : currentOp.dataCount = kernelParam->op.algOperator.dataCount;
453 3 : currentOp.root = kernelParam->op.algOperator.root;
454 3 : currentOp.sendRecvRemoteRank = kernelParam->op.algOperator.sendRecvRemoteRank;
455 :
456 3 : if (kernelParam->op.algOperator.opType != OpType::BATCHSENDRECV) {
457 1 : currentOp.inputMem = std::make_shared<Buffer>(kernelParam->op.input.addr, kernelParam->op.input.size);
458 1 : currentOp.outputMem = std::make_shared<Buffer>(kernelParam->op.output.addr, kernelParam->op.output.size);
459 : }
460 6 : currentOp.scratchMem = std::make_shared<Buffer>(rmaBufferLiteVec[BufferType::SCRATCH]->GetAddr(),
461 6 : rmaBufferLiteVec[BufferType::SCRATCH]->GetSize());
462 3 : kernelParam->op.algOperator.scratchMem = currentOp.scratchMem;
463 3 : if (kernelParam->op.algOperator.scratchMem != nullptr) {
464 9 : HCCL_INFO("CommunicatorImplLite::InitCurrentOp scratchMem addr %llx, size %llu",
465 : kernelParam->op.algOperator.scratchMem->GetAddr(), kernelParam->op.algOperator.scratchMem->GetSize());
466 : }
467 9 : HCCL_INFO("CommunicatorImplLite::InitCurrentOp end");
468 3 : }
469 :
470 : constexpr u32 TAILADDR_OFFSET_MULTIPLIER = 2;
471 0 : void CommunicatorImplLite::SetDfxOpInfo(uint64_t beginTime)
472 : {
473 0 : u64 size = 4;
474 0 : auto dfxopInfo = std::make_shared<DfxOpInfo>();
475 0 : dfxopInfo->op_ = currentOp;
476 0 : dfxopInfo->tag_ = currentOp.opTag;
477 0 : dfxopInfo->algType_ = AlgType{AlgType::MESH}.Describe();
478 0 : dfxopInfo->commIndex_ = idIndex_;
479 0 : dfxopInfo->beginTime_ = beginTime;
480 0 : dfxopInfo->comm_ = this;
481 0 : dfxopInfo->commId_ = commId;
482 0 : dfxopInfo->opIndex_ = opIndex_;
483 0 : dfxopInfo->headOpCounterAddr_ = opCounterAddr + size;
484 0 : dfxopInfo->tailOpCounterAddr_ = opCounterAddr + size * TAILADDR_OFFSET_MULTIPLIER;
485 0 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[SetDfxOpInfo]master stream is nullptr!");
486 0 : mirrorTaskMgrLite->SetCurrDfxOpInfo(dfxopInfo);
487 0 : }
488 :
489 0 : void CommunicatorImplLite::CreateOneSidedComponentLite()
490 : {
491 0 : if (oneSidedComponentLite.get() == nullptr) {
492 : oneSidedComponentLite
493 0 : = make_unique<OneSidedComponentLite>(myRank, rankSize, devType, scratchSize, connectedLinkMgr.get(),
494 0 : rmtDataBufferMgr.get());
495 0 : HCCL_INFO("CommunicatorImplLite::CreateOneSidedComponentLite is null");
496 : }
497 0 : }
498 :
499 0 : std::shared_ptr<InsQueue> CommunicatorImplLite::GetOneSidedInsQueue(HcclKernelParamLite *kernelParam)
500 : {
501 0 : HCCL_INFO("CommunicatorImplLite::GetOneSidedInsQueue begin kernelParam->algName = %s", kernelParam->algName);
502 0 : std::shared_ptr<InsQueue> queue = std::make_shared<InsQueue>();
503 :
504 0 : auto ret = oneSidedComponentLite->Orchestrate(kernelParam->op, queue);
505 0 : if (ret == HCCL_E_PARA) {
506 0 : HCCL_ERROR("CommunicatorImplLite::GetOneSidedInsQueue ret[HCCL_E_PARA]");
507 0 : THROW<InternalException>(StringFormat("CommunicatorImplLite::GetOneSidedInsQueue ret[HCCL_E_PARA]"));
508 : }
509 0 : return queue;
510 0 : }
511 :
512 0 : HcclResult CommunicatorImplLite::SendErrorMessageReportToHost(ErrorMessageReport & errMsgInfo)
513 : {
514 0 : if (kfcStatusTransferD2H == nullptr) {
515 0 : return HCCL_E_PTR;
516 : }
517 0 : CHK_RET(kfcStatusTransferD2H->Put(sizeof(KfcStatus) + sizeof(KfcErrType), sizeof(errMsgInfo),
518 : reinterpret_cast<uint8_t *>(&errMsgInfo)));
519 :
520 0 : return HCCL_SUCCESS;
521 : }
522 :
523 0 : void CommunicatorImplLite::UpdateUserStreamId(HcclKernelParamLite *kernelParam)
524 : {
525 0 : userStreamId_ = kernelParam->op.userStreamId;
526 0 : }
527 :
528 : } // namespace Hccl
|
