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_comm_aicpu.h"
12 : #include "adapter_hal_pub.h"
13 : #include "aicpu_ts_thread.h"
14 : #include "aicpu_res_package_helper.h"
15 : #include "ub_transport_lite_impl.h"
16 : #include "notify_manager.h"
17 : #include "ns_recovery/aicpu/ns_recovery_func_lite.h"
18 : #include "dlhal_function_v2.h"
19 : #include "profiling_command_handle_lite.h"
20 : #include "aicpu_daemon_service.h"
21 : #include "hcclCommTaskExceptionLite.h"
22 : #include "coll_comm_aicpu_destroy_func.h"
23 : #include "aicpu_indop_env.h"
24 : #include "unified_platform/pub_inc/config_plf_log.h"
25 :
26 : constexpr u32 NOTIFY_SIZE_EIGHT = 8;
27 :
28 20 : CollCommAicpu::~CollCommAicpu()
29 : {
30 20 : std::unique_lock<std::shared_mutex> rwLock(threadMutex_);
31 22 : for (auto& thread : threads_) {
32 2 : HcommThreadRegisterCheckExecStatus(reinterpret_cast<ThreadHandle>(thread.get()), nullptr);
33 : }
34 20 : threads_.clear();
35 20 : HCCL_RUN_INFO("[CollCommAicpu][%s]Group[%s] destroy success", __func__, identifier_.c_str());
36 20 : }
37 :
38 0 : HcclResult CollCommAicpu::InitAicpuIndOp(CommAicpuParam *commAicpuParam)
39 : {
40 0 : if (commStatus_ == HcclCommStatus::HCCL_COMM_STATUS_READY) {
41 0 : HCCL_RUN_INFO("[CollCommAicpu][%s]Group[%s] already initialized, skip reinit", __func__,
42 : identifier_.c_str());
43 0 : return HCCL_SUCCESS;
44 : }
45 0 : CHK_PTR_NULL(commAicpuParam);
46 0 : topoInfo_.deviceLogicId = commAicpuParam->deviceLogicId;
47 0 : topoInfo_.devicePhyId = commAicpuParam->devicePhyId;
48 0 : topoInfo_.deviceType = static_cast<DevType>(commAicpuParam->deviceType);
49 0 : identifier_ = std::string(commAicpuParam->hcomId);
50 0 : topoInfo_.userRankSize = commAicpuParam->userRankSize;
51 0 : topoInfo_.userRank = commAicpuParam->userRank;
52 0 : notifys_.reserve(hccl::HCCL_THREAD_NOTIFY_MAX_NUM);
53 :
54 0 : CHK_RET(hrtSetWorkModeAicpu(true));
55 0 : CHK_RET(hrtSetlocalDevice(topoInfo_.deviceLogicId));
56 0 : CHK_RET(hrtSetlocalDeviceType(topoInfo_.deviceType));
57 0 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(topoInfo_.devicePhyId, &devId_));
58 0 : CHK_RET(dfx_.Init(devId_, identifier_, topoInfo_.userRankSize));
59 0 : CHK_RET(RegisterProfCallBack());
60 0 : CHK_RET(InitHDCommunicate(commAicpuParam));
61 :
62 0 : EXCEPTION_CATCH(nsRecoveryLitePtr_ = std::make_shared<NsRecoveryLite>(), return HCCL_E_PTR);
63 0 : nsRecoveryLitePtr_->Init(kfcControlTransferH2D_, kfcStatusTransferD2H_);
64 :
65 0 : CHK_RET(Hccl::DlHalFunctionV2::GetInstance().DlHalFunctionInit());
66 :
67 0 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
68 :
69 : static std::once_flag initBackGround;
70 0 : std::call_once(initBackGround, [this]() { this->InitBackGroundThread();} );
71 :
72 : static std::once_flag initEnv;
73 0 : std::call_once(initEnv, [this, commAicpuParam]() { this->InitIndopEnv(commAicpuParam);} );
74 0 : HCCL_RUN_INFO("[%s]success, group[%s], deviceLogicId[%u], devicePhyId[%u], deviceType[%u], rankSize[%u] "\
75 : "userRank[%u], devId[%u]", __func__, identifier_.c_str(), topoInfo_.deviceLogicId, topoInfo_.devicePhyId,
76 : topoInfo_.deviceType, topoInfo_.userRankSize, topoInfo_.userRank, devId_);
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : HcclResult CollCommAicpu::InitHDCommunicate(CommAicpuParam *commAicpuParam)
81 : {
82 0 : if (commAicpuParam->kfcControlTransferH2DParams.buffLen != 0 && kfcControlTransferH2D_ == nullptr) {
83 0 : EXCEPTION_CATCH((kfcControlTransferH2D_ = std::make_shared<hccl::HDCommunicate>()), return HCCL_E_PTR);
84 0 : CHK_SMART_PTR_NULL(kfcControlTransferH2D_);
85 0 : CHK_RET(kfcControlTransferH2D_->InitDevice(commAicpuParam->kfcControlTransferH2DParams));
86 : }
87 0 : if (commAicpuParam->kfcStatusTransferD2HParams.buffLen != 0 && kfcStatusTransferD2H_ == nullptr) {
88 0 : EXCEPTION_CATCH((kfcStatusTransferD2H_ = std::make_shared<hccl::HDCommunicate>()), return HCCL_E_PTR);
89 0 : CHK_SMART_PTR_NULL(kfcStatusTransferD2H_);
90 0 : CHK_RET(kfcStatusTransferD2H_->InitDevice(commAicpuParam->kfcStatusTransferD2HParams));
91 : }
92 0 : return HCCL_SUCCESS;
93 : }
94 :
95 0 : void CollCommAicpu::InitIndopEnv(CommAicpuParam *commAicpuParam)
96 : {
97 0 : hcomm::SetTaskExceptionEnable(commAicpuParam->commConfig.taskExceptionEnable);
98 0 : Hccl::SetPlfDebugConfigValue(commAicpuParam->commConfig.plfDebugConfig);
99 0 : HCCL_RUN_INFO("[%s]Env: taskExceptionEnable[%d], notifyWaitTimeout[%u], plfDebugConfig[0x%llx]",
100 : __func__, commAicpuParam->commConfig.taskExceptionEnable, commAicpuParam->commConfig.notifyWaitTimeout,
101 : commAicpuParam->commConfig.plfDebugConfig);
102 0 : }
103 :
104 7 : void CollCommAicpu::SetCommmStatus(HcclCommStatus status)
105 : {
106 7 : HCCL_INFO("[%s]group[%s], flag[%d]", __func__, identifier_.c_str(), static_cast<int>(status));
107 7 : commStatus_ = status;
108 7 : }
109 :
110 1 : HcclResult CollCommAicpu::InitThreads(ThreadMgrAicpuParam *param)
111 : {
112 1 : u32 threadNum = param->threadNum;
113 1 : std::vector<std::shared_ptr<Thread>> outThreads;
114 1 : outThreads.reserve(threadNum);
115 1 : std::string hcomId(param->hcomId);
116 2 : for (u32 i = 0; i < threadNum; ++i) {
117 1 : std::string thdUniqueId(param->threadParam[i], THREAD_UNIQUE_ID_MAX_SIZE);
118 1 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
119 1 : std::ostringstream oss;
120 1 : oss << "threadParam[" << i << "] raw bytes: ";
121 6001 : for (u32 j = 0; j < THREAD_UNIQUE_ID_MAX_SIZE; ++j) {
122 6000 : oss << std::hex << std::setw(2) << std::setfill('0')
123 6000 : << static_cast<unsigned int>(static_cast<unsigned char>(param->threadParam[i][j])) << " ";
124 : }
125 1 : HCCL_INFO("[CollCommAicpu][%s] %s", __func__, oss.str().c_str());
126 1 : }
127 1 : std::shared_ptr<AicpuTsThread> thread;
128 1 : EXCEPTION_CATCH((thread = std::make_shared<AicpuTsThread>(thdUniqueId)), return HCCL_E_PTR);
129 1 : HcclResult ret = thread->Init();
130 1 : if (ret != HCCL_SUCCESS) {
131 0 : HCCL_ERROR("[CollCommAicpu][%s] comm identifier[%s], init threads num[%u] failed at index %u",
132 : __func__, hcomId.c_str(), param->threadNum, i);
133 0 : return ret;
134 : }
135 1 : outThreads.emplace_back(thread);
136 1 : }
137 :
138 1 : ThreadHandle *threadArray = static_cast<ThreadHandle*>(param->deviceHandle);
139 : // 空指针校验
140 1 : CHK_PTR_NULL(threadArray);
141 2 : for (size_t i = 0; i < outThreads.size(); ++i) {
142 1 : threadArray[i] = reinterpret_cast<ThreadHandle>(outThreads[i].get()); // 拷贝裸指针
143 1 : HCCL_INFO("[CollCommAicpu][%s] threadArray[%u] = [%lu]", __func__, i, threadArray[i]);
144 1 : CHK_RET(RegisterThreadAddDfxTaskInfo(threadArray[i]));
145 : }
146 1 : std::unique_lock<std::shared_mutex> rwLock(threadMutex_);
147 1 : threads_.insert(threads_.end(), std::make_move_iterator(outThreads.begin()),
148 : std::make_move_iterator(outThreads.end()));
149 1 : HCCL_INFO("[CollCommAicpu][%s] comm identifier[%s], init threads num[%u] success",
150 : __func__, hcomId.c_str(), threadNum);
151 1 : return HCCL_SUCCESS;
152 1 : }
153 :
154 0 : HcclResult CollCommAicpu::RegisterThreadAddDfxTaskInfo(ThreadHandle thread)
155 : {
156 0 : int32_t ret = HcommThreadRegisterDfx(thread, dfx_.GetCallback());
157 0 : if (ret != 0) {
158 0 : HCCL_ERROR("[%s] HcommThreadRegisterDfx failed, ret[%d], thread[0x%llx], dfx_.GetCallback[%p]",
159 : __func__, ret, thread, dfx_.GetCallback());
160 0 : return HCCL_E_PTR;
161 : }
162 :
163 0 : std::function<HcclResult(bool)> checkExecStatusCallback = [this](bool isTimeout) {
164 0 : return this->CheckIndOpExecStatus(isTimeout);
165 0 : };
166 0 : ret = HcommThreadRegisterCheckExecStatus(thread, checkExecStatusCallback);
167 0 : if (ret != 0) {
168 0 : HCCL_ERROR("[%s]HcommThreadRegisterCheckExecStatus failed, ret[%d], thread[0x%llx], checkExecStatusCallback[%p]",
169 : __func__, ret, thread, checkExecStatusCallback);
170 0 : return HCCL_E_PTR;
171 : }
172 0 : return HCCL_SUCCESS;
173 0 : }
174 :
175 0 : HcclResult CollCommAicpu::AllocChannelResource(HcclChannelUrmaRes *commParam)
176 : {
177 0 : HCCL_INFO("[CollCommAicpu][%s] deviceLogicId[%d], devicePhyId[%u], deviceType[%d], commParam->channelList[%p], "
178 : "commParam->listNum[%u], commParam->uniqueIdAddr[%p], commParam->uniqueIdSize[%u]",
179 : __func__, topoInfo_.deviceLogicId, topoInfo_.devicePhyId, topoInfo_.deviceType, commParam->channelList,
180 : commParam->listNum, commParam->uniqueIdAddr, commParam->uniqueIdSize);
181 0 : CHK_RET(InitUrmaChannel(commParam));
182 0 : return HCCL_SUCCESS;
183 : }
184 :
185 0 : HcclResult CollCommAicpu::ProcessUrmaRes(HcclChannelUrmaRes *commParam, bool isInit)
186 : {
187 0 : HCCL_INFO("[CollCommAicpu][%s] commParam->uniqueIdAddr[%p], commParam->uniqueIdSize[%u]",
188 : __func__, commParam->uniqueIdAddr, commParam->uniqueIdSize);
189 0 : ChannelHandle* channelList = reinterpret_cast<ChannelHandle*>(commParam->channelList);
190 0 : u8* currentSrcAddr = reinterpret_cast<u8*>(commParam->uniqueIdAddr);
191 0 : u32* addSize = reinterpret_cast<u32*>(commParam->channelSizeAddr);
192 0 : CHK_PTR_NULL(channelList);
193 0 : CHK_PTR_NULL(currentSrcAddr);
194 0 : CHK_PTR_NULL(addSize);
195 :
196 0 : for (u32 index = 0; index < commParam->listNum; index++) {
197 0 : std::vector<char> data(*addSize);
198 :
199 : // 计算地址块的偏移
200 0 : CHK_SAFETY_FUNC_RET(memcpy_s(data.data(), data.size(), currentSrcAddr, *addSize));
201 0 : currentSrcAddr += *addSize;
202 0 : addSize++;
203 : // 反序列化得到device侧transport对象
204 : Hccl::AicpuResPackageHelper helper;
205 0 : auto dataVec = helper.ParsePackedData(data);
206 :
207 0 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM; // 待修改
208 0 : if (static_cast<u32>(resType) >= dataVec.size()) {
209 0 : HCCL_ERROR("[CollCommAicpu][%s] fail, resType[%d], dataVec size[%u]", __func__, resType, dataVec.size());
210 0 : return HCCL_E_PARA;
211 : }
212 :
213 0 : ChannelHandle channelHandle{0};
214 0 : if (isInit) {
215 0 : CHK_RET(ParsePackData(dataVec[resType].data, channelHandle));
216 : // 恢复出的channelHandle回填到commParam中
217 0 : channelList[index] = channelHandle;
218 0 : CHK_RET(RegisterChannelAddDfxTaskInfo(channelHandle));
219 0 : dfx_.AddChannelRemoteRankId(channelHandle, commParam->remoteRankList[index]);
220 : } else {
221 0 : channelHandle = channelList[index];
222 0 : if (!ubTransportMap_.count(channelHandle)) {
223 0 : HCCL_ERROR("[CollCommAicpu][%s] fail, resType[%d], current ChannelHandle nullptr", __func__, resType);
224 0 : return HCCL_E_PARA;
225 : }
226 0 : CHK_RET(ResumePackData(dataVec[resType].data, channelHandle));
227 : }
228 :
229 : // 打印
230 0 : HCCL_INFO("[CollCommAicpu][%s] index[%u], currentSrcAddr[%p], channelSizeAddr[%p], channelHandle[0x%llx]",
231 : __func__, index, currentSrcAddr, commParam->channelSizeAddr, channelHandle);
232 0 : }
233 :
234 0 : return HCCL_SUCCESS;
235 : }
236 :
237 0 : HcclResult CollCommAicpu::InitUrmaChannel(HcclChannelUrmaRes *commParam)
238 : {
239 0 : return ProcessUrmaRes(commParam, true);
240 : }
241 :
242 0 : HcclResult CollCommAicpu::ParsePackData(std::vector<char> &data, ChannelHandle &handle)
243 : {
244 0 : HCCL_DEBUG("[CollCommAicpu][%s] data: ptr[%p], size[%u]", __func__, data.data(), data.size());
245 0 : Hccl::BinaryStream binaryStream(data);
246 :
247 0 : std::vector<char> transpUniqueId;
248 0 : binaryStream >> transpUniqueId;
249 :
250 0 : Hccl::BinaryStream binaryStreamForType(transpUniqueId);
251 : u32 transType;
252 0 : binaryStreamForType >> transType;
253 0 : HCCL_INFO("[CollCommAicpu][ParsePackData] transType[%u]", transType);
254 0 : if (transType == Hccl::TransportType::UB) {
255 0 : std::unique_ptr<Hccl::UbTransportLiteImpl> ubTransportLiteImpl;
256 0 : EXCEPTION_CATCH((ubTransportLiteImpl = std::make_unique<Hccl::UbTransportLiteImpl>(transpUniqueId)),
257 : return HCCL_E_PTR);
258 0 : CHK_SMART_PTR_NULL(ubTransportLiteImpl);
259 0 : ubTransportLiteImpl->SetTaskExceptionEnable(hcomm::GetTaskExceptionEnable());
260 0 : handle = reinterpret_cast<uint64_t>(ubTransportLiteImpl.get());
261 0 : ubTransportMap_.insert({handle, std::move(ubTransportLiteImpl)});
262 0 : } else if (transType == Hccl::TransportType::P2P) {
263 0 : std::unique_ptr<Hccl::P2PTransportLiteImpl> p2pTransportLiteImpl;
264 0 : EXCEPTION_CATCH((p2pTransportLiteImpl = std::make_unique<Hccl::P2PTransportLiteImpl>(transpUniqueId)),
265 : return HCCL_E_PTR);
266 0 : CHK_SMART_PTR_NULL(p2pTransportLiteImpl);
267 0 : handle = reinterpret_cast<uint64_t>(p2pTransportLiteImpl.get());
268 0 : p2pTransportMap_.insert({handle, std::move(p2pTransportLiteImpl)});
269 0 : } else if (transType == Hccl::TransportType::ROCE) {
270 0 : std::unique_ptr<Hccl::RoceTransportLiteImpl> roceTransportLiteImpl;
271 0 : EXCEPTION_CATCH((roceTransportLiteImpl = std::make_unique<Hccl::RoceTransportLiteImpl>(transpUniqueId)),
272 : return HCCL_E_PTR);
273 0 : CHK_SMART_PTR_NULL(roceTransportLiteImpl);
274 :
275 0 : handle = reinterpret_cast<uint64_t>(roceTransportLiteImpl.get());
276 0 : roceTransportMap_.insert({handle, std::move(roceTransportLiteImpl)});
277 0 : } else {
278 0 : HCCL_ERROR("[CollCommAicpu][ParsePackData] unsupported transportType[%u]", transType);
279 0 : return HCCL_E_INTERNAL;
280 : }
281 :
282 0 : return HCCL_SUCCESS;
283 0 : }
284 :
285 0 : HcclResult CollCommAicpu::RegisterChannelAddDfxTaskInfo(ChannelHandle channel) {
286 0 : int hert = HcommChannelRegisterDfx(channel, dfx_.GetCallback());
287 0 : return static_cast<HcclResult>(hert);
288 : }
289 :
290 0 : HcclResult CollCommAicpu::NotifyFree(NotifyMgrAicpuParam *param)
291 : {
292 0 : u32 notifyNum = param->notifyNum;
293 0 : NotifyHandle *notifyArray = static_cast<NotifyHandle*>(param->deviceHandle);
294 0 : std::string hcomId(param->hcomId);
295 : // 空指针校验
296 0 : CHK_PTR_NULL(notifyArray);
297 0 : for (size_t i = 0; i < notifyNum; ++i) {
298 0 : LocalNotify* notify = reinterpret_cast<LocalNotify*>(notifyArray[i]);
299 0 : HCCL_INFO("[CollCommAicpu][%s] notifyArray[%u]=[%lu]", __func__, i, notifyArray[i]);
300 0 : auto it = std::find_if(notifys_.begin(), notifys_.end(),
301 0 : [notify](const std::unique_ptr<LocalNotify>& ptr) {
302 0 : return ptr.get() == notify;
303 : });
304 0 : if (it != notifys_.end()) {
305 0 : HCCL_INFO("[CollCommAicpu][%s] comm identifier[%s], free notifys[%u] success",
306 : __func__, hcomId.c_str(), notifyArray[i]);
307 0 : notifys_.erase(it);
308 : } else {
309 0 : HCCL_RUN_WARNING("[CollCommAicpu][%s] localNotify[%u] not found in notifys_", __func__, i);
310 : }
311 : }
312 :
313 0 : HCCL_INFO("[CollCommAicpu][%s] comm identifier[%s], free notifys num[%u] success",
314 : __func__, hcomId.c_str(), notifyNum);
315 0 : return HCCL_SUCCESS;
316 0 : }
317 :
318 0 : HcclResult CollCommAicpu::NotifyAlloc(NotifyMgrAicpuParam *param)
319 : {
320 0 : u32 notifyNum = param->notifyNum;
321 0 : std::string notifysStr = std::string(param->notifyParam, NOTIFY_UNIQUE_ID_MAX_SIZE);
322 0 : std::string hcomId(param->hcomId);
323 0 : size_t notifySize = notifys_.size();
324 0 : HCCL_INFO("[CollCommAicpu][%s] comm identifier[%s], alloc notifys num[%u] begin, before notifySize[%u]",
325 : __func__, hcomId.c_str(), notifyNum, notifySize);
326 0 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
327 0 : std::ostringstream oss;
328 0 : oss << "notifyParam" << " raw bytes: ";
329 0 : for (u32 i = 0; i < NOTIFY_UNIQUE_ID_MAX_SIZE; ++i) {
330 0 : oss << std::hex << std::setw(2) << std::setfill('0')
331 0 : << static_cast<unsigned int>(static_cast<unsigned char>(param->notifyParam[i])) << " ";
332 : }
333 0 : HCCL_INFO("[CollCommAicpu][%s] %s", __func__, oss.str().c_str());
334 0 : }
335 0 : HcclResult ret = NotifyManager::ParseBinNotifys(notifysStr, notifys_);
336 0 : if (ret != HCCL_SUCCESS) {
337 0 : HCCL_ERROR("[CollCommAicpu][%s] comm identifier[%s], alloc notifys num[%u] failed %u",
338 : __func__, hcomId.c_str(), notifyNum, ret);
339 0 : return ret;
340 : }
341 0 : HCCL_INFO("[CollCommAicpu][%s] comm identifier[%s], alloc notifys num[%u] end, after notifySize[%u]",
342 : __func__, hcomId.c_str(), notifyNum, notifys_.size());
343 0 : NotifyHandle *notifyArray = static_cast<NotifyHandle*>(param->deviceHandle);
344 0 : CHK_PTR_NULL(notifyArray);
345 : // 空指针校验
346 0 : for (size_t i = 0; i < notifyNum; ++i) {
347 0 : notifyArray[i] = reinterpret_cast<NotifyHandle>(notifys_[i + notifySize].get()); // 拷贝裸指针
348 0 : HCCL_INFO("[CollCommAicpu][%s] notifyArray[%u] = [%lu]", __func__, i + notifySize, notifyArray[i]);
349 : }
350 :
351 0 : HCCL_INFO("[CollCommAicpu][%s] comm identifier[%s], alloc notifys num[%u] success",
352 : __func__, hcomId.c_str(), notifyNum);
353 0 : return HCCL_SUCCESS;
354 0 : }
355 :
356 3 : hccl::NsRecoveryLitePtr CollCommAicpu::GetNsRecoveryLitePtr()
357 : {
358 3 : return nsRecoveryLitePtr_;
359 : }
360 :
361 1 : HcclResult CollCommAicpu::Clean()
362 : {
363 1 : for (auto& transPort : ubTransportMap_) {
364 0 : CHK_RET(transPort.second->Clean());
365 : }
366 1 : HCCL_INFO("CollCommAicpu::Clean() finished");
367 :
368 1 : return HCCL_SUCCESS;
369 : }
370 :
371 0 : HcclResult CollCommAicpu::ResumePackData(std::vector<char> &data, ChannelHandle &handle)
372 : {
373 0 : Hccl::BinaryStream binaryStream(data);
374 0 : std::vector<char> transpUniqueId;
375 0 : binaryStream >> transpUniqueId;
376 :
377 0 : auto& transPortPtr = ubTransportMap_[handle];
378 0 : CHK_RET(transPortPtr->Resume(transpUniqueId));
379 0 : return HCCL_SUCCESS;
380 0 : }
381 :
382 1 : HcclResult CollCommAicpu::Resume(HcclChannelUrmaRes *commParam)
383 : {
384 1 : CHK_PTR_NULL(commParam);
385 1 : CHK_RET(ProcessUrmaRes(commParam, false));
386 1 : nsRecoveryLitePtr_->SetNeedClean(false);
387 :
388 1 : SetErrorReported(false);
389 1 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
390 :
391 1 : return HCCL_SUCCESS;
392 : }
393 :
394 1 : HcclResult CollCommAicpu::CheckIndOpExecStatus(bool timeout)
395 : {
396 1 : if (timeout) {
397 0 : HCCL_ERROR("[%s]comm[%s] op launch timeout, print taskException", __func__, identifier_.c_str());
398 : // 先打印本通信域的taskException，再打印其他通信域的taskException
399 0 : hcomm::HcclCommTaskExceptionLite::GetInstance().PrintCommTaskException(this);
400 0 : hcomm::HcclCommTaskExceptionLite::GetInstance().PrintAllCommTaskException();
401 0 : return HCCL_E_INTERNAL;
402 1 : } else if (commStatus_ == HCCL_COMM_STATUS_SUSPENDING) {
403 1 : HCCL_WARNING("[%s]comm[%s] commStatus[%d] is suspending", __func__, identifier_.c_str(), commStatus_);
404 1 : return HCCL_E_SUSPENDING;
405 0 : } else if (commStatus_ != HCCL_COMM_STATUS_READY) {
406 0 : HCCL_ERROR("[%s]comm[%s] commStatus[%d] is not ready, return fail", __func__, identifier_.c_str(), commStatus_);
407 0 : return HCCL_E_INTERNAL;
408 : }
409 0 : return HCCL_SUCCESS;
410 : }
411 :
412 0 : void CollCommAicpu::InitBackGroundThread()
413 : {
414 : static auto commandToBackGroud = Hccl::CommandToBackGroud::Default;
415 0 : static auto daemonServiceRun = [](void *info) {
416 0 : Hccl::AicpuDaemonService::GetInstance().ServiceRun(info);
417 0 : };
418 0 : static auto daemonServiceStop = [](void *info) {
419 0 : Hccl::AicpuDaemonService::GetInstance().ServiceStop(info);
420 0 : };
421 :
422 : // 注册守护进程函数
423 0 : hcomm::HcclCommTaskExceptionLite::GetInstance().Init(devId_);
424 0 : Hccl::AicpuDaemonService::GetInstance().Register(&hcomm::HcclCommTaskExceptionLite::GetInstance());
425 0 : Hccl::AicpuDaemonService::GetInstance().Register(&hccl::CollCommAicpuDestroyFunc::GetInstance());
426 0 : Hccl::AicpuDaemonService::GetInstance().Register(&NsRecoveryFuncLite::GetInstance());
427 :
428 : // 启动背景线程
429 0 : if (Hccl::StartMC2MaintenanceThread != nullptr) {
430 0 : Hccl::StartMC2MaintenanceThread(daemonServiceRun, &commandToBackGroud, daemonServiceStop, &commandToBackGroud);
431 0 : HCCL_RUN_INFO("[%s]start BackGround thread success.", __func__);
432 : } else {
433 0 : HCCL_WARNING("[%s]StartMC2MaintenanceThread func is nullptr", __func__);
434 : }
435 0 : }
436 :
437 0 : HcclResult CollCommAicpu::BackGroundGetCmd(Hccl::KfcCommand &cmd)
438 : {
439 0 : CHK_SMART_PTR_NULL(kfcControlTransferH2D_);
440 0 : HcclResult ret = kfcControlTransferH2D_->Get(0, sizeof(Hccl::KfcCommand), reinterpret_cast<uint8_t *>(&cmd));
441 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]fail, group[%s]", __func__, identifier_.c_str()), ret);
442 0 : return HCCL_SUCCESS;
443 : }
444 :
445 0 : HcclResult CollCommAicpu::BackGroundSetStatus(Hccl::KfcStatus state)
446 : {
447 0 : Hccl::KfcExecStatus status;
448 0 : status.kfcStatus = state;
449 0 : HCCL_INFO("[%s]group[%s], state[%u]", __func__, identifier_.c_str(), state);
450 0 : HcclResult ret = kfcStatusTransferD2H_->Put(0, sizeof(status.kfcStatus), reinterpret_cast<uint8_t *>(&status));
451 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]fail, group[%s]", __func__, identifier_.c_str()), ret);
452 0 : return HCCL_SUCCESS;
453 : }
454 :
455 0 : HcclResult CollCommAicpu::SendErrorMessageReportToHost(Hccl::ErrorMessageReport& errMsgInfo)
456 : {
457 0 : CHK_SMART_PTR_NULL(kfcStatusTransferD2H_);
458 0 : CHK_RET(kfcStatusTransferD2H_->Put(sizeof(Hccl::KfcStatus) + sizeof(Hccl::KfcErrType), sizeof(errMsgInfo),
459 : reinterpret_cast<uint8_t *>(&errMsgInfo)));
460 0 : return HCCL_SUCCESS;
461 : }
462 :
463 0 : HcclResult CollCommAicpu::RegisterProfCallBack()
464 : {
465 0 : if (MsprofRegisterCallback != nullptr) {
466 0 : HCCL_INFO("RegisterProfCallBack not null");
467 0 : int32_t ret = MsprofRegisterCallback(AICPU, &Hccl::DeviceCommandHandle);
468 0 : CHK_PRT_RET((ret != 0), HCCL_ERROR("[%s] failed. ret = [%d]", __func__, ret), HCCL_E_PARA);
469 : } else {
470 0 : HCCL_INFO("RegisterProfCallBack is null");
471 : }
472 0 : return HCCL_SUCCESS;
473 : }
474 :
475 0 : u32 CollCommAicpu::UpdateIndex()
476 : {
477 0 : return index_+=1;
478 : }
|
