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 "coll_comm_aicpu_mgr.h"
13 : #include "aicpu_communicator.h"
14 : #include "adapter_hal_pub.h"
15 : #include "ns_recovery/aicpu/ns_recovery_func_lite.h"
16 : #include "dlhal_function_v2.h"
17 : #include "dfx_profiling_command_handle_lite.h"
18 : #include "hcclCommTaskExceptionLite.h"
19 : #include "hcclCommOp.h"
20 : #include "hcclCommDfxLite.h"
21 : #include "env_config/env_config_v2.h"
22 : #include "log.h"
23 : #include <shared_mutex>
24 :
25 54 : CollCommAicpu::~CollCommAicpu()
26 : {
27 54 : HCCL_RUN_INFO("[CollCommAicpu][%s]Group[%s] destroy success", __func__, identifier_.c_str());
28 54 : }
29 :
30 1 : HcclResult CollCommAicpu::InitAicpuIndOp(CommAicpuParam* commAicpuParam)
31 : {
32 1 : if (commStatus_ == HcclCommStatus::HCCL_COMM_STATUS_READY) {
33 0 : HCCL_RUN_INFO("[CollCommAicpu][%s]Group[%s] already initialized, skip reinit", __func__, identifier_.c_str());
34 0 : return HCCL_SUCCESS;
35 : }
36 1 : CHK_PTR_NULL(commAicpuParam);
37 1 : topoInfo_.deviceLogicId = commAicpuParam->deviceLogicId;
38 1 : topoInfo_.devicePhyId = commAicpuParam->devicePhyId;
39 1 : topoInfo_.deviceType = static_cast<DevType>(commAicpuParam->deviceType);
40 1 : identifier_ = std::string(commAicpuParam->hcomId);
41 1 : topoInfo_.userRankSize = commAicpuParam->userRankSize;
42 1 : topoInfo_.userRank = commAicpuParam->userRank;
43 :
44 1 : CHK_RET(hrtSetWorkModeAicpu(true));
45 1 : CHK_RET(hrtSetlocalDevice(topoInfo_.deviceLogicId));
46 1 : CHK_RET(hrtSetlocalDeviceType(topoInfo_.deviceType));
47 1 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(topoInfo_.devicePhyId, &devId_));
48 1 : CHK_RET(dfx_.Init(devId_, identifier_, topoInfo_.userRankSize, topoInfo_.userRank));
49 1 : CHK_RET(RegisterProfCallBack());
50 1 : CHK_RET(InitHDCommunicate(commAicpuParam));
51 :
52 1 : EXCEPTION_CATCH(nsRecoveryLitePtr_ = std::make_shared<NsRecoveryLite>(), return HCCL_E_PTR);
53 1 : nsRecoveryLitePtr_->Init(kfcControlTransferH2D_, kfcStatusTransferD2H_);
54 :
55 1 : CHK_RET(Hccl::DlHalFunctionV2::GetInstance().DlHalFunctionInit());
56 :
57 : // commEngineResMgr_/channelMgr_ 为 CollCommAicpu 成员（unique_ptr），生命周期被 this 严格包含，
58 : // 因此 lambda 捕获 this 安全，不会产生悬垂指针（析构顺序见 coll_comm_aicpu.h 成员声明）
59 1 : EXCEPTION_CATCH(
60 : commEngineResMgr_ = std::make_unique<CommEngineResAicpuMgr>(
61 : dfx_,
62 : [this](bool isTimeout) {
63 : return this->CheckIndOpExecStatus(isTimeout);
64 : }),
65 : return HCCL_E_PTR);
66 1 : CHK_SMART_PTR_NULL(commEngineResMgr_);
67 :
68 1 : EXCEPTION_CATCH(channelMgr_ = std::make_unique<ChannelAicpuMgr>(dfx_, topoInfo_), return HCCL_E_PTR);
69 1 : CHK_SMART_PTR_NULL(channelMgr_);
70 :
71 1 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
72 :
73 1 : HCCL_RUN_INFO(
74 : "[%s]success, group[%s], deviceLogicId[%u], devicePhyId[%u], deviceType[%u], rankSize[%u] "
75 : "userRank[%u], devId[%u]",
76 : __func__, identifier_.c_str(), topoInfo_.deviceLogicId, topoInfo_.devicePhyId, topoInfo_.deviceType,
77 : topoInfo_.userRankSize, topoInfo_.userRank, devId_);
78 1 : return HCCL_SUCCESS;
79 : }
80 :
81 1 : HcclResult CollCommAicpu::InitHDCommunicate(CommAicpuParam* commAicpuParam)
82 : {
83 1 : if (commAicpuParam->kfcControlTransferH2DParams.buffLen != 0 && kfcControlTransferH2D_ == nullptr) {
84 0 : EXCEPTION_CATCH((kfcControlTransferH2D_ = std::make_shared<hccl::HDCommunicate>()), return HCCL_E_PTR);
85 0 : CHK_SMART_PTR_NULL(kfcControlTransferH2D_);
86 0 : CHK_RET(kfcControlTransferH2D_->InitDevice(commAicpuParam->kfcControlTransferH2DParams));
87 : }
88 1 : if (commAicpuParam->kfcStatusTransferD2HParams.buffLen != 0 && kfcStatusTransferD2H_ == nullptr) {
89 0 : EXCEPTION_CATCH((kfcStatusTransferD2H_ = std::make_shared<hccl::HDCommunicate>()), return HCCL_E_PTR);
90 0 : CHK_SMART_PTR_NULL(kfcStatusTransferD2H_);
91 0 : CHK_RET(kfcStatusTransferD2H_->InitDevice(commAicpuParam->kfcStatusTransferD2HParams));
92 : }
93 1 : return HCCL_SUCCESS;
94 : }
95 :
96 13 : void CollCommAicpu::SetCommmStatus(HcclCommStatus status)
97 : {
98 13 : HCCL_INFO("[%s]group[%s], commStatus[%d]", __func__, identifier_.c_str(), static_cast<int>(status));
99 13 : commStatus_ = status;
100 13 : }
101 :
102 1 : HcclResult CollCommAicpu::Clean()
103 : {
104 1 : CHK_SMART_PTR_NULL(channelMgr_);
105 0 : return channelMgr_->Clean();
106 : }
107 :
108 1 : HcclResult CollCommAicpu::Resume(HcclChannelUrmaRes* commParam)
109 : {
110 1 : CHK_PTR_NULL(commParam);
111 1 : CHK_SMART_PTR_NULL(channelMgr_);
112 1 : CHK_RET(channelMgr_->Resume(commParam));
113 1 : nsRecoveryLitePtr_->SetNeedClean(false);
114 :
115 : // 重置Rtsq维护的 sq tail/head
116 1 : std::shared_lock<std::shared_mutex> lock(GetCommEngineResMgr()->GetThreadMutex());
117 1 : const std::vector<std::shared_ptr<hccl::Thread>>& threads = GetCommEngineResMgr()->GetAllThread();
118 1 : for (auto& thread : threads) {
119 0 : Hccl::StreamLite* streamLitePtr = reinterpret_cast<Hccl::StreamLite*>(thread->GetStreamLitePtr());
120 0 : streamLitePtr->GetRtsq()->Reset(true);
121 : }
122 :
123 1 : SetErrorReported(false);
124 1 : commStatus_ = HcclCommStatus::HCCL_COMM_STATUS_READY;
125 1 : HCCL_RUN_INFO("[CollCommAicpu][Resume] commId[%s] resume success.", identifier_.c_str());
126 :
127 1 : return HCCL_SUCCESS;
128 1 : }
129 :
130 3 : hccl::NsRecoveryLitePtr CollCommAicpu::GetNsRecoveryLitePtr() { return nsRecoveryLitePtr_; }
131 :
132 2 : HcclResult CollCommAicpu::CheckIndOpExecStatus(bool timeout)
133 : {
134 2 : if (timeout) {
135 0 : HCCL_ERROR("[%s]comm[%s] op launch timeout, print taskException", __func__, identifier_.c_str());
136 0 : hcomm::HcclCommTaskExceptionLite::GetInstance().PrintCommTaskException(this);
137 0 : hcomm::HcclCommTaskExceptionLite::GetInstance().PrintAllCommTaskException();
138 0 : return HCCL_E_INTERNAL;
139 2 : } else if (commStatus_ == HCCL_COMM_STATUS_SUSPENDING) {
140 1 : HCCL_WARNING("[%s]comm[%s] commStatus[%d] is suspending", __func__, identifier_.c_str(), commStatus_);
141 1 : return HCCL_E_SUSPENDING;
142 1 : } else if (commStatus_ != HCCL_COMM_STATUS_READY) {
143 1 : HCCL_ERROR("[%s]comm[%s] commStatus[%d] is not ready, return fail", __func__, identifier_.c_str(), commStatus_);
144 1 : return HCCL_E_INTERNAL;
145 : }
146 0 : return HCCL_SUCCESS;
147 : }
148 :
149 0 : HcclResult CollCommAicpu::BackGroundGetCmd(Hccl::KfcCommand& cmd)
150 : {
151 0 : CHK_SMART_PTR_NULL(kfcControlTransferH2D_);
152 0 : HcclResult ret = kfcControlTransferH2D_->Get(0, sizeof(Hccl::KfcCommand), reinterpret_cast<uint8_t*>(&cmd));
153 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]fail, group[%s]", __func__, identifier_.c_str()), ret);
154 0 : return HCCL_SUCCESS;
155 : }
156 :
157 0 : HcclResult CollCommAicpu::BackGroundSetStatus(Hccl::KfcStatus state)
158 : {
159 0 : Hccl::KfcExecStatus status;
160 0 : status.kfcStatus = state;
161 0 : HCCL_INFO("[%s]group[%s], state[%d]", __func__, identifier_.c_str(), static_cast<int>(state));
162 0 : HcclResult ret = kfcStatusTransferD2H_->Put(0, sizeof(status.kfcStatus), reinterpret_cast<uint8_t*>(&status));
163 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[%s]fail, group[%s]", __func__, identifier_.c_str()), ret);
164 0 : return HCCL_SUCCESS;
165 : }
166 :
167 0 : HcclResult CollCommAicpu::SendErrorMessageReportToHost(Hccl::ErrorMessageReport& errMsgInfo)
168 : {
169 0 : CHK_SMART_PTR_NULL(kfcStatusTransferD2H_);
170 0 : CHK_RET(kfcStatusTransferD2H_->Put(
171 : sizeof(Hccl::KfcStatus) + sizeof(Hccl::KfcErrType), sizeof(errMsgInfo),
172 : reinterpret_cast<uint8_t*>(&errMsgInfo)));
173 0 : return HCCL_SUCCESS;
174 : }
175 :
176 1 : HcclResult CollCommAicpu::RegisterProfCallBack() { return Hccl::DfxRegisterProfCallBack(); }
177 :
178 2 : u32 CollCommAicpu::UpdateIndex() { return index_ += 1; }
179 :
180 0 : HcclResult CollCommAicpu::InitDfxOpInfo(HcclDfxOpInfo* aicpuDfxInfo)
181 : {
182 0 : HCCL_INFO(
183 : "[%s]group[%s], algTag[%s], profiling L0[%d], L1[%d]", __func__, identifier_.c_str(), aicpuDfxInfo->algTag,
184 : Hccl::DfxProfilingHandlerLite::GetInstance().GetProfL0State(),
185 : Hccl::DfxProfilingHandlerLite::GetInstance().GetProfL1State());
186 :
187 0 : Hccl::DfxDfxOpInfo newDfxOpInfo{};
188 0 : newDfxOpInfo.opType = static_cast<u8>(aicpuDfxInfo->opType);
189 0 : newDfxOpInfo.dataType = static_cast<u8>(aicpuDfxInfo->dataType);
190 :
191 0 : newDfxOpInfo.commHandle = reinterpret_cast<void*>(this);
192 0 : newDfxOpInfo.count = aicpuDfxInfo->dataCount;
193 0 : newDfxOpInfo.srcAddr = aicpuDfxInfo->inputMemAddr;
194 0 : newDfxOpInfo.dstAddr = aicpuDfxInfo->outputMemAddr;
195 0 : newDfxOpInfo.srcSize = aicpuDfxInfo->inputMemSize;
196 0 : newDfxOpInfo.dstSize = aicpuDfxInfo->outputMemSize;
197 0 : newDfxOpInfo.opIndex = UpdateIndex();
198 0 : newDfxOpInfo.cpuWaitAicpuNotifyId = aicpuDfxInfo->cpuWaitAicpuNotifyId;
199 0 : newDfxOpInfo.algType = static_cast<u8>(Hccl::AlgTypeVal::ALG_TYPE_NOT_SPECIFIED);
200 0 : auto algTagLen = strnlen(aicpuDfxInfo->algTag, sizeof(newDfxOpInfo.algTag) - 1);
201 0 : CHK_SAFETY_FUNC_RET(
202 : memcpy_s(newDfxOpInfo.algTag, sizeof(newDfxOpInfo.algTag) - 1, aicpuDfxInfo->algTag, algTagLen));
203 :
204 0 : CHK_RET(dfx_.SetCurrDfxOpInfo(&newDfxOpInfo));
205 0 : return HCCL_SUCCESS;
206 : }
207 :
208 0 : HcclResult CollCommAicpu::ProfilingReportDeviceOp()
209 : {
210 0 : HcclCommDfxLite* hcclCommDfxLite = GetHcclCommDfxLite();
211 0 : CHK_PTR_NULL(hcclCommDfxLite);
212 0 : auto* currDfxOpInfo = static_cast<const Hccl::DfxDfxOpInfo*>(hcclCommDfxLite->GetLatestDfxOpInfo());
213 0 : if (currDfxOpInfo == nullptr) {
214 0 : HCCL_WARNING("[%s] no op info registered, skip ProfilingReportDeviceOp.", __func__);
215 0 : return HCCL_SUCCESS;
216 : }
217 :
218 0 : const auto& sharedThreads = commEngineResMgr_->GetAllThread();
219 0 : std::vector<hccl::Thread*> threads;
220 0 : threads.reserve(sharedThreads.size());
221 0 : for (const auto& t : sharedThreads) {
222 0 : threads.push_back(t.get());
223 : }
224 0 : hcclCommDfxLite->ReportAllTasks(threads);
225 0 : EXCEPTION_CATCH(
226 : Hccl::DfxProfilingHandlerLite::GetInstance().ReportHcclOpInfo(*currDfxOpInfo), return HCCL_E_INTERNAL);
227 0 : return HCCL_SUCCESS;
228 0 : }
229 :
230 0 : HcclResult CollCommAicpu::UpdateTask()
231 : {
232 0 : CHK_RET(dfx_.UpdateProfStat());
233 0 : return HCCL_SUCCESS;
234 : }
235 :
236 51 : hccl::HcclCommAicpu* CollCommAicpu::GetLegacy910CollComm() { return legacy910CollComm_.first.get(); }
237 :
238 11 : void CollCommAicpu::SetLegacy910CollComm(std::shared_ptr<hccl::HcclCommAicpu> comm)
239 : {
240 11 : legacy910CollComm_.first = std::move(comm);
241 11 : }
242 :
243 3 : bool CollCommAicpu::IsLegacy910CollCommBusy() { return legacy910CollComm_.second.load(); }
244 :
245 59 : void CollCommAicpu::SetLegacy910CollCommBusy(bool busy) { legacy910CollComm_.second.store(busy); }
|
