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 0 : algComponentLite = make_unique<CollAlgComponentLite>(
112 0 : myRank, rankSize, devType, scratchSize, connectedLinkMgr.get(), rmtDataBufferMgr.get());
113 0 : HCCL_INFO("CommunicatorImplLite::CreateCollAlgComponentLite is null");
114 : } else {
115 0 : algComponentLite->UpdateScratchBufferSize(scratchSize);
116 0 : HCCL_INFO("CommunicatorImplLite::CreateCollAlgComponentLite, bufferSize %llu", scratchSize);
117 : }
118 0 : }
119 :
120 0 : void CommunicatorImplLite::UnfoldOp(HcclKernelParamLite* kernelParam)
121 : {
122 0 : opIndex_ = kernelParam->comm.opIndex_;
123 0 : uint64_t beginTime = ProfGetCurCpuTimestamp();
124 0 : profilingReporterLite->UpdateProfStat();
125 0 : UpdateCommParam(kernelParam);
126 0 : UpdateLocBuffer(kernelParam);
127 0 : UpdateUserStreamId(kernelParam);
128 0 : UpdateRes(kernelParam);
129 0 : UpdateDynamicOpData(kernelParam);
130 0 : SetDfxOpInfo(beginTime);
131 :
132 0 : UpdateHDCommnicate(kernelParam);
133 0 : RegisterRtsqCallback();
134 :
135 0 : isCommReady = true;
136 0 : HCCL_INFO("CommunicatorImplLite::UnfoldOpBase isCommReady is set to true.");
137 0 : std::shared_ptr<InsQueue> insQueue = GetInsQueue(kernelParam);
138 0 : if (insQueue == nullptr) {
139 0 : THROW<NullPtrException>(StringFormat("CommunicatorImplLite::UnfoldOpBase insQueue is nullptr."));
140 : }
141 0 : if (devType == DevType::DEV_TYPE_950 || devType == DevType::DEV_TYPE_960) {
142 0 : HCCL_INFO("CommunicatorImplLite::UnfoldOpBase DevType is DEV_TYPE_950 or DEV_TYPE_960.");
143 0 : insExecutor->ExecuteV82(*insQueue);
144 0 : profilingReporterLite->ReportAllTasks();
145 0 : ProfilingHandlerLite::GetInstance().ReportHcclOpInfo(*mirrorTaskMgrLite->GetCurrDfxOpInfo());
146 0 : } else if (devType == DevType::DEV_TYPE_910A2) {
147 0 : HCCL_INFO("CommunicatorImplLite::UnfoldOpBase DevType is DEV_TYPE_910A2.");
148 0 : insExecutor->Execute(*insQueue);
149 : } else {
150 0 : HCCL_WARNING("CommunicatorImplLite::UnfoldOpBase DevType is not support.");
151 : }
152 0 : kernelParam->op.algOperator.scratchMem = nullptr;
153 0 : }
154 :
155 0 : void CommunicatorImplLite::RegisterRtsqCallback()
156 : {
157 0 : auto checkOpExecStatusCallback = [this]() {
158 0 : this->CheckOpExecStatus();
159 0 : };
160 0 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[RegisterRtsqCallback]master stream is nullptr!");
161 0 : streamLiteMgr->GetMaster()->GetRtsq()->SetOpExecStatusCallback(checkOpExecStatusCallback);
162 0 : for (u32 i = 0; i < streamLiteMgr->SizeOfSlaves(); ++i) {
163 0 : streamLiteMgr->GetSlave(i)->GetRtsq()->SetOpExecStatusCallback(checkOpExecStatusCallback);
164 : }
165 0 : }
166 :
167 0 : void CommunicatorImplLite::CheckOpExecStatus() const
168 : {
169 0 : if (isSuspended) {
170 0 : HCCL_INFO("hccl aicpu stop wait finish, for recv stop launch cmd");
171 0 : THROW<SuspendingException>(
172 0 : StringFormat("[CheckOpExecStatus] recv stop launch command, coll service is suspended."));
173 : }
174 0 : }
175 :
176 5 : void CommunicatorImplLite::UpdateCommParam(HcclKernelParamLite* kernelParam)
177 : {
178 5 : if (isUpdateComm) {
179 0 : return;
180 : }
181 5 : myRank = kernelParam->comm.myRank;
182 5 : rankSize = kernelParam->comm.rankSize;
183 5 : devPhyId = kernelParam->comm.devPhyId;
184 5 : devType = kernelParam->comm.devType;
185 5 : opCounterAddr = kernelParam->comm.opCounterAddr;
186 5 : hcclExecTimeout = kernelParam->envConfig.hcclExecTimeout;
187 5 : if (rmtDataBufferMgr == nullptr) {
188 4 : collAlgInfo = std::make_unique<CollAlgInfo>(kernelParam->op.algOperator.opMode, kernelParam->opTag);
189 4 : rmtDataBufferMgr = std::make_unique<RmtDataBufferMgr>(transportLiteMgr.get(), collAlgInfo.get());
190 : }
191 5 : commId = kernelParam->comm.commId;
192 15 : HCCL_INFO(
193 : "CommunicatorImplLite::UpdateCommParam myRank [%u] rankSize[%u] devPhyId[%u] devType[%d] scratchSize [%llu] "
194 : "scratchaddress[%llx] opCounterAddr[%llx] commId[%s]",
195 : myRank, rankSize, devPhyId, devType, scratchSize, kernelParam->comm.opBaseScratch.addr, opCounterAddr,
196 : commId.c_str());
197 5 : isUpdateComm = true;
198 : }
199 :
200 3 : void CommunicatorImplLite::UpdateLocBuffer(HcclKernelParamLite* kernelParam)
201 : {
202 3 : locBuffer[BufferType::INPUT] = kernelParam->op.input.addr;
203 3 : locBuffer[BufferType::OUTPUT] = kernelParam->op.output.addr;
204 :
205 3 : rmaBufferLiteVec.clear();
206 3 : rmaBufferLiteVec.resize(BufferType::__COUNT__);
207 :
208 3 : InitRmaBufferLite(kernelParam->op.input, BufferType::INPUT);
209 3 : InitRmaBufferLite(kernelParam->op.output, BufferType::OUTPUT);
210 :
211 3 : if (kernelParam->op.algOperator.opMode == OpMode::OPBASE) {
212 0 : scratchSize = kernelParam->comm.opBaseScratch.size;
213 0 : locBuffer[BufferType::SCRATCH] = kernelParam->comm.opBaseScratch.addr;
214 0 : InitRmaBufferLite(kernelParam->comm.opBaseScratch, BufferType::SCRATCH);
215 : } else {
216 3 : scratchSize = kernelParam->op.scratch.size;
217 3 : locBuffer[BufferType::SCRATCH] = kernelParam->op.scratch.addr;
218 3 : InitRmaBufferLite(kernelParam->op.scratch, BufferType::SCRATCH);
219 : }
220 :
221 3 : if (kernelParam->oneSidedComm) {
222 0 : if ((kernelParam->op.batchPutGetLocalAddr == nullptr) || (kernelParam->op.batchPutGetRemoteAddr == nullptr)) {
223 0 : THROW<InternalException>("batchPutGetAddr is nullptr");
224 : }
225 : }
226 :
227 3 : InitCurrentOp(kernelParam);
228 9 : HCCL_INFO(
229 : "CommunicatorImplLite::UpdateLocBuffer locBuffer[BufferType::INPUT] %llx, locBuffer[BufferType::OUTPUT] %llx",
230 : locBuffer[BufferType::INPUT], locBuffer[BufferType::OUTPUT]);
231 9 : HCCL_INFO(
232 : "CommunicatorImplLite::UpdateLocBuffer locBuffer[BufferType::SCRATCH] %llx", locBuffer[BufferType::SCRATCH]);
233 3 : }
234 :
235 1 : void CommunicatorImplLite::UpdateTransports(HcclKernelParamLite* kernelParam)
236 : {
237 3 : HCCL_INFO("[NsRecovery] RestoreAllTransports start");
238 1 : RestoreAllTransports(kernelParam->binaryResAddr, kernelParam->binaryResSize);
239 3 : HCCL_INFO("[NsRecovery] RestoreAllTransports end");
240 1 : }
241 :
242 1 : void CommunicatorImplLite::RestoreAllTransports(u64 addr, u64 bufSize)
243 : {
244 1 : std::vector<char> data;
245 1 : data.resize(bufSize);
246 1 : int ret = memcpy_s(data.data(), bufSize, reinterpret_cast<void*>(addr), bufSize);
247 1 : if (ret != 0) {
248 0 : THROW<InternalException>(
249 0 : StringFormat("[NsRecovery] CommunicatorImplLite::RestoreAllTransports: memcpy_s failed, ret = %d", ret));
250 : }
251 3 : HCCL_INFO(
252 : "[NsRecovery] CommunicatorImplLite::RestoreAllTransports: RestoreData %s",
253 : Bytes2hex(data.data(), data.size()).c_str());
254 : AicpuResPackageHelper helper;
255 1 : auto dataVec = helper.ParsePackedData(data);
256 :
257 1 : AicpuResMgrType resType = AicpuResMgrType::TRANSPORT;
258 1 : GetTransportLiteMgr()->ParseAllPackedData(dataVec[resType].data);
259 3 : HCCL_INFO("[NsRecovery] CommunicatorImplLite::RestoreAllTransports: GetResMgr %s Data", resType.Describe().c_str());
260 1 : }
261 :
262 0 : bool CommunicatorImplLite::CheckNeedUpdateRes(HcclKernelParamLite* kernelParam)
263 : {
264 0 : std::string tagKey = kernelParam->tagKey;
265 0 : auto it = loadedOpSet.find(tagKey);
266 0 : if (it != loadedOpSet.end()) {
267 0 : HCCL_INFO("[CheckNeedUpdateRes] Corresponding resources of tag[%s] have been loaded", tagKey.c_str());
268 0 : return false;
269 : }
270 0 : loadedOpSet.insert(tagKey);
271 0 : return true;
272 0 : }
273 :
274 0 : void CommunicatorImplLite::UpdateRes(HcclKernelParamLite* kernelParam)
275 : {
276 0 : if (CheckNeedUpdateRes(kernelParam)) {
277 0 : HCCL_INFO("[UpdateRes] start, opMode[%s]", kernelParam->op.algOperator.opMode.Describe().c_str());
278 0 : RestoreOpRes(kernelParam->opTag, kernelParam->tagKey, kernelParam->binaryResAddr, kernelParam->binaryResSize);
279 0 : HCCL_INFO("[UpdateRes] end");
280 : }
281 0 : }
282 :
283 0 : void CommunicatorImplLite::UpdateHDCommnicate(HcclKernelParamLite* kernelParam)
284 : {
285 0 : CHK_RET_THROW(
286 : InternalException,
287 : StringFormat("[CommunicatorImplLite][%s] failed to init kfcControlTransferH2DParams", __func__),
288 : kfcControlTransferH2D->Init(kernelParam->kfcControlTransferH2DParams));
289 0 : CHK_RET_THROW(
290 : InternalException,
291 : StringFormat("[CommunicatorImplLite][%s] failed to init kfcControlTransferD2HParams", __func__),
292 : kfcStatusTransferD2H->Init(kernelParam->kfcControlTransferD2HParams));
293 0 : std::unique_lock<std::mutex> lock(hdcShmLock_);
294 0 : hdcHandler = make_unique<AicpuHdcHandler>(*kfcControlTransferH2D, *kfcStatusTransferD2H);
295 0 : }
296 :
297 0 : void CommunicatorImplLite::UpdateDynamicOpData(HcclKernelParamLite* kernelParam) const
298 : {
299 0 : HCCL_INFO(
300 : "[CommunicatorImplLite][UpdateDynamicOpData] OpType[%s]",
301 : kernelParam->op.algOperator.opType.Describe().c_str());
302 0 : u8* dynamicDataPtr = reinterpret_cast<u8*>(kernelParam) + sizeof(struct HcclKernelParamLite);
303 0 : if (kernelParam->op.algOperator.opType == OpType::BATCHSENDRECV) {
304 0 : struct BatchSendRecvDataDes* batchSendRecvDataPtr
305 : = reinterpret_cast<struct BatchSendRecvDataDes*>(dynamicDataPtr);
306 0 : kernelParam->op.algOperator.batchSendRecvDataDes.itemNum = batchSendRecvDataPtr->itemNum;
307 0 : kernelParam->op.algOperator.batchSendRecvDataDes.sendRecvItemsPtr = batchSendRecvDataPtr->batchSendRecvItem;
308 0 : HcclSendRecvItem* itemPtr
309 : = reinterpret_cast<HcclSendRecvItem*>(kernelParam->op.algOperator.batchSendRecvDataDes.sendRecvItemsPtr);
310 0 : for (u32 i = 0; i < kernelParam->op.algOperator.batchSendRecvDataDes.itemNum; i++) {
311 0 : HCCL_INFO(
312 : "[CommunicatorImplLite][UpdateDynamicOpData] batchSendRecvDataDes remoteRank[%u]",
313 : (itemPtr + i)->remoteRank);
314 : }
315 0 : } else if (kernelParam->op.algOperator.opType == OpType::ALLTOALLV) {
316 0 : struct AllToAllvDataDes* alltoallvDataPtr = reinterpret_cast<struct AllToAllvDataDes*>(dynamicDataPtr);
317 0 : kernelParam->op.algOperator.all2AllVDataDes.sendType = static_cast<DataType::Value>(alltoallvDataPtr->sendType);
318 0 : kernelParam->op.algOperator.all2AllVDataDes.recvType = static_cast<DataType::Value>(alltoallvDataPtr->recvType);
319 0 : u64 rankSize = kernelParam->comm.rankSize;
320 0 : kernelParam->op.algOperator.all2AllVDataDes.sendCounts = static_cast<void*>(alltoallvDataPtr->sendRecvInfos);
321 : kernelParam->op.algOperator.all2AllVDataDes.recvCounts
322 0 : = static_cast<void*>(static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + rankSize);
323 0 : kernelParam->op.algOperator.all2AllVDataDes.sdispls = static_cast<void*>(
324 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_DATA_INDEX_2 * rankSize);
325 0 : kernelParam->op.algOperator.all2AllVDataDes.rdispls = static_cast<void*>(
326 0 : static_cast<u64*>(alltoallvDataPtr->sendRecvInfos) + ALLTOALLV_DATA_INDEX_3 * rankSize);
327 0 : for (u32 i = 0; i < rankSize; i++) {
328 0 : HCCL_INFO(
329 : "[CommunicatorImplLite][UpdateDynamicOpData] alltoallv sendCounts[%llu], recvCounts[%llu]",
330 : *(static_cast<const u64*>(kernelParam->op.algOperator.all2AllVDataDes.sendCounts) + i),
331 : *(static_cast<const u64*>(kernelParam->op.algOperator.all2AllVDataDes.recvCounts) + i));
332 : }
333 0 : } else if (kernelParam->op.algOperator.opType == OpType::ALLTOALLVC) {
334 0 : struct AllToAllvcDataDes* alltoallvcDataPtr = reinterpret_cast<struct AllToAllvcDataDes*>(dynamicDataPtr);
335 : kernelParam->op.algOperator.all2AllVCDataDes.sendType
336 0 : = static_cast<DataType::Value>(alltoallvcDataPtr->sendType);
337 : kernelParam->op.algOperator.all2AllVCDataDes.recvType
338 0 : = static_cast<DataType::Value>(alltoallvcDataPtr->recvType);
339 : kernelParam->op.algOperator.all2AllVCDataDes.sendCountMatrix
340 0 : = static_cast<void*>(alltoallvcDataPtr->sendCountMatrix);
341 : }
342 0 : }
343 :
344 21 : HostDeviceSyncNotifyLiteMgr* CommunicatorImplLite::GetHostDeviceSyncNotifyLiteMgr()
345 : {
346 21 : return hostDeviceSyncNotifyLiteMgr.get();
347 : }
348 :
349 50 : StreamLiteMgr* CommunicatorImplLite::GetStreamLiteMgr() { return streamLiteMgr.get(); }
350 :
351 1 : QueueNotifyLiteMgr* CommunicatorImplLite::GetQueueNotifyLiteMgr() { return queueNotifyLiteMgr.get(); }
352 :
353 1 : Cnt1tonNotifyLiteMgr* CommunicatorImplLite::GetCnt1tonNotifyLiteMgr() { return cnt1tonNotifyLiteMgr.get(); }
354 :
355 1 : CntNto1NotifyLiteMgr* CommunicatorImplLite::GetCntNto1NotifyLiteMgr() { return cntNto1NotifyLiteMgr.get(); }
356 :
357 1 : ConnectedLinkMgr* CommunicatorImplLite::GetConnectedLinkMgr() { return connectedLinkMgr.get(); }
358 :
359 89 : DevId CommunicatorImplLite::GetDevPhyId() { return devPhyId; }
360 :
361 0 : u32 CommunicatorImplLite::GetExecTimeOut() { return hcclExecTimeout; }
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(
393 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
394 :
395 0 : resType = AicpuResMgrType::QUEUE_NOTIFY;
396 0 : GetQueueNotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
397 0 : HCCL_INFO(
398 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
399 :
400 0 : resType = AicpuResMgrType::QUEUE_WAIT_GROUP_CNT_NOTIFY;
401 0 : GetCntNto1NotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
402 0 : HCCL_INFO(
403 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
404 :
405 0 : resType = AicpuResMgrType::QUEUE_BCAST_POST_CNT_NOTIFY;
406 0 : GetCnt1tonNotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
407 0 : HCCL_INFO(
408 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
409 :
410 0 : resType = AicpuResMgrType::HOST_DEV_SYNC_NOTIFY;
411 0 : GetHostDeviceSyncNotifyLiteMgr()->ParsePackedData(dataVec[resType].data);
412 0 : HCCL_INFO(
413 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
414 :
415 0 : resType = AicpuResMgrType::TRANSPORT;
416 0 : if (currentOp.opMode == OpMode::OPBASE) { // 单算子模式
417 0 : GetTransportLiteMgr()->ParseOpbasePackedData(dataVec[resType].data);
418 0 : } else if (currentOp.opMode == OpMode::OFFLOAD) { // 图下沉模式
419 0 : GetTransportLiteMgr()->ParseOffloadPackedData(opTag, dataVec[resType].data);
420 : } else {
421 0 : THROW<InternalException>(StringFormat("opMode=%s failed", currentOp.opMode.Describe().c_str()));
422 : }
423 0 : HCCL_INFO(
424 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data, %s", opTag.c_str(), resType.Describe().c_str(),
425 : currentOp.opMode.Describe().c_str());
426 :
427 0 : resType = AicpuResMgrType::ALG_TOPO;
428 : AlgTopoPackageHelper algTopoHelper;
429 0 : algTopoInfoMap[tagKey] = algTopoHelper.GetAlgTopoInfo(dataVec[resType].data);
430 0 : HCCL_INFO(
431 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data, tagKey=%s", opTag.c_str(),
432 : resType.Describe().c_str(), tagKey.c_str());
433 :
434 0 : resType = AicpuResMgrType::CONNECTD_MGR;
435 0 : GetConnectedLinkMgr()->ParsePackedData(dataVec[resType].data);
436 0 : HCCL_INFO(
437 : "CommunicatorImplLite::RestoreOpRes: opTag %s GetResMgr %s Data", opTag.c_str(), resType.Describe().c_str());
438 0 : }
439 :
440 80 : CommunicatorImplLite::CommunicatorImplLite(u32 idIndex) : idIndex_(idIndex) {}
441 :
442 9 : void CommunicatorImplLite::InitRmaBufferLite(HcclAicpuLocBufLite& bufLite, BufferType type)
443 : {
444 9 : rmaBufferLiteVec[type]
445 18 : = std::make_unique<RmaBufferLite>(bufLite.addr, bufLite.size, bufLite.tokenId, bufLite.tokenValue);
446 9 : }
447 :
448 3 : void CommunicatorImplLite::InitCurrentOp(HcclKernelParamLite* kernelParam)
449 : {
450 3 : currentOp.opTag = kernelParam->opTag; // opTag的赋值
451 9 : HCCL_INFO("CommunicatorImplLite::InitCurrentOp opTag[%s]", currentOp.opTag.c_str());
452 :
453 3 : currentOp.opMode = kernelParam->op.algOperator.opMode;
454 3 : currentOp.opType = kernelParam->op.algOperator.opType;
455 3 : currentOp.reduceOp = kernelParam->op.algOperator.reduceOp;
456 3 : currentOp.dataType = kernelParam->op.algOperator.dataType;
457 3 : currentOp.outputDataType = kernelParam->op.algOperator.outputDataType;
458 3 : currentOp.dataCount = kernelParam->op.algOperator.dataCount;
459 3 : currentOp.root = kernelParam->op.algOperator.root;
460 3 : currentOp.sendRecvRemoteRank = kernelParam->op.algOperator.sendRecvRemoteRank;
461 :
462 3 : if (kernelParam->op.algOperator.opType != OpType::BATCHSENDRECV) {
463 1 : currentOp.inputMem = std::make_shared<Buffer>(kernelParam->op.input.addr, kernelParam->op.input.size);
464 1 : currentOp.outputMem = std::make_shared<Buffer>(kernelParam->op.output.addr, kernelParam->op.output.size);
465 : }
466 3 : currentOp.scratchMem = std::make_shared<Buffer>(
467 3 : rmaBufferLiteVec[BufferType::SCRATCH]->GetAddr(), rmaBufferLiteVec[BufferType::SCRATCH]->GetSize());
468 3 : kernelParam->op.algOperator.scratchMem = currentOp.scratchMem;
469 3 : if (kernelParam->op.algOperator.scratchMem != nullptr) {
470 9 : HCCL_INFO(
471 : "CommunicatorImplLite::InitCurrentOp scratchMem addr %llx, size %llu",
472 : kernelParam->op.algOperator.scratchMem->GetAddr(), kernelParam->op.algOperator.scratchMem->GetSize());
473 : }
474 9 : HCCL_INFO("CommunicatorImplLite::InitCurrentOp end");
475 3 : }
476 :
477 : constexpr u32 TAILADDR_OFFSET_MULTIPLIER = 2;
478 0 : void CommunicatorImplLite::SetDfxOpInfo(uint64_t beginTime)
479 : {
480 0 : u64 size = 4;
481 0 : auto dfxopInfo = std::make_shared<DfxOpInfo>();
482 0 : dfxopInfo->op_ = currentOp;
483 0 : dfxopInfo->tag_ = currentOp.opTag;
484 0 : dfxopInfo->algType_ = AlgType{AlgType::MESH}.Describe();
485 0 : dfxopInfo->commIndex_ = idIndex_;
486 0 : dfxopInfo->beginTime_ = beginTime;
487 0 : dfxopInfo->comm_ = this;
488 0 : dfxopInfo->commId_ = commId;
489 0 : dfxopInfo->opIndex_ = opIndex_;
490 0 : dfxopInfo->headOpCounterAddr_ = opCounterAddr + size;
491 0 : dfxopInfo->tailOpCounterAddr_ = opCounterAddr + size * TAILADDR_OFFSET_MULTIPLIER;
492 0 : CHECK_NULLPTR(streamLiteMgr->GetMaster(), "[SetDfxOpInfo]master stream is nullptr!");
493 0 : mirrorTaskMgrLite->SetCurrDfxOpInfo(dfxopInfo);
494 0 : }
495 :
496 0 : void CommunicatorImplLite::CreateOneSidedComponentLite()
497 : {
498 0 : if (oneSidedComponentLite.get() == nullptr) {
499 0 : oneSidedComponentLite = make_unique<OneSidedComponentLite>(
500 0 : myRank, rankSize, devType, scratchSize, connectedLinkMgr.get(), rmtDataBufferMgr.get());
501 0 : HCCL_INFO("CommunicatorImplLite::CreateOneSidedComponentLite is null");
502 : }
503 0 : }
504 :
505 0 : std::shared_ptr<InsQueue> CommunicatorImplLite::GetOneSidedInsQueue(HcclKernelParamLite* kernelParam)
506 : {
507 0 : HCCL_INFO("CommunicatorImplLite::GetOneSidedInsQueue begin kernelParam->algName = %s", kernelParam->algName);
508 0 : std::shared_ptr<InsQueue> queue = std::make_shared<InsQueue>();
509 :
510 0 : auto ret = oneSidedComponentLite->Orchestrate(kernelParam->op, queue);
511 0 : if (ret == HCCL_E_PARA) {
512 0 : HCCL_ERROR("CommunicatorImplLite::GetOneSidedInsQueue ret[HCCL_E_PARA]");
513 0 : THROW<InternalException>(StringFormat("CommunicatorImplLite::GetOneSidedInsQueue ret[HCCL_E_PARA]"));
514 : }
515 0 : return queue;
516 0 : }
517 :
518 0 : HcclResult CommunicatorImplLite::SendErrorMessageReportToHost(ErrorMessageReport& errMsgInfo)
519 : {
520 0 : if (kfcStatusTransferD2H == nullptr) {
521 0 : return HCCL_E_PTR;
522 : }
523 0 : CHK_RET(kfcStatusTransferD2H->Put(
524 : sizeof(KfcStatus) + sizeof(KfcErrType), sizeof(errMsgInfo), reinterpret_cast<uint8_t*>(&errMsgInfo)));
525 :
526 0 : return HCCL_SUCCESS;
527 : }
528 :
529 0 : void CommunicatorImplLite::UpdateUserStreamId(HcclKernelParamLite* kernelParam)
530 : {
531 0 : userStreamId_ = kernelParam->op.userStreamId;
532 0 : }
533 :
534 : } // namespace Hccl
|
