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 <mutex>
12 : #include <shared_mutex>
13 : #include <iterator>
14 : #include <functional>
15 : #include <map>
16 : #include "framework/aicpu_hccl_process.h"
17 : #include "stream_pub.h"
18 : #include "framework/aicpu_communicator.h"
19 : #include "sal_pub.h"
20 : #include "cann_error_reporter.h"
21 : #include "executor_tracer.h"
22 : #include "dfx/aicpu_executor_tracer.h"
23 : #include "framework/aicpu_one_side_service.h"
24 :
25 : namespace dfx_tracer {
26 140 : void ExecutorTracer::BackGroundDfx(void* info)
27 : {
28 140 : HCCL_RUN_INFO("Start to back ground.");
29 : // 外部保证info有效
30 140 : auto ctx = static_cast<AicpuComContext*>(info);
31 140 : hccl::HcclCommAicpu::ResetErrMsgReport(); // 业务重新拉起的场景，重置ErrMesg上报标记位
32 : while (true) {
33 : // 停止背景线程
34 8240 : if (ctx->dfxExtendInfo.commandToBackGroud == CommandToBackGroud::kStop) {
35 139 : HCCL_INFO("Back ground thread returned.");
36 139 : break;
37 : }
38 8101 : HandleDestroyComm(ctx);
39 8101 : HandleBackGround(ctx);
40 8101 : bool isNotStop = false;
41 8101 : StopBackGround(ctx, isNotStop);
42 8101 : if (!isNotStop) {
43 1 : HCCL_RUN_INFO("stop backGround Thread");
44 1 : break;
45 : }
46 8100 : HandleReportStatusInComm();
47 8100 : StopLaunchCommandHandle(ctx);
48 8100 : KfcCommandHandle(ctx);
49 8100 : HandleSwitchNic(ctx);
50 8100 : TaskMonitor();
51 8100 : HandleCqeStatus(ctx);
52 8100 : HandleResumeChangeLink(ctx);
53 8100 : hccl::HcclOneSideServiceAicpu::HandleErrCqe();
54 8100 : usleep(TEN_MILLISECOND_OF_USLEEP);
55 8100 : }
56 140 : (void)dfx::CannErrorReporter::GetInstance().Clear();
57 140 : }
58 :
59 8101 : void ExecutorTracer::HandleDestroyComm(AicpuComContext* const ctx)
60 : {
61 8101 : std::vector<std::string> destroyGroupName;
62 : {
63 8101 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
64 8101 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
65 8101 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
66 8101 : KfcCommand cmd = KfcCommand::kNone;
67 8101 : for (auto& commInfo : aicpuCommInfo) {
68 0 : hccl::HcclCommAicpu* hcclAicpu = commInfo.second;
69 0 : if (hcclAicpu->GetCommInfoStatus() || hcclAicpu->GetIsInitIndOp()) {
70 0 : (void)hcclAicpu->GetKfcCommand(cmd);
71 0 : if (cmd == KfcCommand::kDestroyComm) {
72 0 : auto groupName = hcclAicpu->GetGroupName();
73 0 : HCCL_RUN_INFO(
74 : "[ExecutorTracer][%s]Recv kDestroyComm cmd, group name[%s]", __func__, groupName.c_str());
75 0 : hcclAicpu->FlushUtraceInfo();
76 0 : KfcExecStatus responseStatus;
77 0 : responseStatus.execStatus.kfcStatus = KfcStatus::kDestroyComm;
78 : // 需要在销毁通信域前返回 kfc status到host，销毁通信域会释放TransferD2H
79 0 : s32 ret = hcclAicpu->ResponseBackGroundStatus(responseStatus);
80 0 : CHK_PRT_CONT(
81 : ret, HCCL_ERROR(
82 : "[ExecutorTracer][%s]ResponseBackGroundStatus failed, group[%s], ret[%d]", __func__,
83 : groupName.c_str(), ret));
84 0 : AicpuExecutorTracer::StopKfcThread(ctx, aicpuCommInfo);
85 0 : destroyGroupName.push_back(groupName);
86 0 : cmd = KfcCommand::kNone;
87 0 : }
88 : }
89 : }
90 8101 : }
91 :
92 8101 : for (auto& groupName : destroyGroupName) {
93 : // DestroyComm 内部已加锁，此处无需再加锁（避免 std::shared_mutex 非递归重复加锁死锁）
94 0 : AicpuHcclProcess::AicpuDestoryCommbyGroup(groupName);
95 : }
96 8101 : }
97 :
98 8100 : void ExecutorTracer::TaskMonitor(void)
99 : {
100 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
101 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
102 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
103 8100 : for (auto& commInfo : aicpuCommInfo) {
104 0 : hccl::HcclCommAicpu* hcclAicpu = commInfo.second;
105 0 : (void)hcclAicpu->StreamTaskMonitor();
106 : }
107 8100 : }
108 :
109 8101 : void ExecutorTracer::HandleBackGround(AicpuComContext* const ctx) { AicpuExecutorTracer::HandleBackGround(ctx); }
110 :
111 : // stop 背景线程
112 8101 : void ExecutorTracer::StopBackGround(AicpuComContext* const ctx, bool& isNotStop)
113 : {
114 8101 : if (ctx->commOpenStatus) {
115 8100 : isNotStop = true;
116 : } else {
117 1 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
118 1 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
119 1 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
120 1 : for (auto& commInfo : aicpuCommInfo) {
121 0 : hccl::HcclCommAicpu* hcclAicpu = commInfo.second;
122 0 : if (hcclAicpu->GetCommInfoStatus() || hcclAicpu->GetIsInitIndOp()) {
123 0 : isNotStop = true;
124 : }
125 : }
126 1 : }
127 :
128 8101 : if (!hccl::HcclOneSideServiceAicpu::isAllDestroy()) {
129 0 : isNotStop = true;
130 : }
131 8101 : }
132 :
133 140 : void ExecutorTracer::StopBackGroundDfx(void* info)
134 : {
135 : // 外部保证info有效
136 140 : auto ctx = static_cast<AicpuComContext*>(info);
137 140 : ctx->dfxExtendInfo.commandToBackGroud = CommandToBackGroud::kStop;
138 140 : HCCL_INFO("Stop back ground thread.");
139 140 : }
140 :
141 : // handle StopLaunch Command
142 8100 : void ExecutorTracer::StopLaunchCommandHandle(AicpuComContext* const ctx)
143 : {
144 8100 : AicpuExecutorTracer::StopLaunchCommandHandle(ctx);
145 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
146 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
147 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
148 8100 : KfcCommand cmd = KfcCommand::kNone;
149 8100 : for (auto& commInfo : aicpuCommInfo) {
150 0 : hccl::HcclCommAicpu* hcclAicpu = commInfo.second;
151 0 : cmd = KfcCommand::kNone;
152 0 : if (hcclAicpu->GetCommInfoStatus()) {
153 0 : if (!hcclAicpu->GetNsStopLaunchStatus()) {
154 0 : (void)hcclAicpu->BackGroundGetCmd(cmd);
155 0 : if (cmd == KfcCommand::NsStopLaunch) {
156 0 : if (!hcclAicpu->BackGroundGetOpStatus()) {
157 0 : (void)hcclAicpu->BackGroundSetStatus(KfcStatus::kStoplaunch);
158 0 : hcclAicpu->SetCommRecoveryFlag(true);
159 0 : hcclAicpu->SetNsStopLaunchStatus(true);
160 0 : HCCL_RUN_INFO(
161 : "[NsRecovery][backGround]group[%s] send in aicpu environment",
162 : hcclAicpu->GetGroupName().c_str());
163 : }
164 : }
165 : }
166 : }
167 : }
168 8100 : }
169 :
170 : // handle StopExec and Clean Command
171 8100 : void ExecutorTracer::KfcCommandHandle(AicpuComContext* const ctx)
172 : {
173 8100 : AicpuExecutorTracer::KfcCommandHandle(ctx);
174 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
175 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
176 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
177 8100 : for (auto& commItem : aicpuCommInfo) {
178 0 : hccl::HcclCommAicpu* commInfo = commItem.second;
179 0 : if (commInfo->GetCommInfoStatus()) {
180 0 : if (commInfo->GetCommRecoveryFlag()) {
181 0 : HandleAICPUCommand(commInfo);
182 : }
183 : }
184 : }
185 8100 : }
186 :
187 : // handle switch nic command
188 8100 : void ExecutorTracer::HandleSwitchNic([[maybe_unused]] AicpuComContext* const ctx)
189 : {
190 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
191 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
192 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
193 8100 : for (auto& commItem : aicpuCommInfo) {
194 0 : hccl::HcclCommAicpu* hcclAicpu = commItem.second;
195 0 : if (hcclAicpu->GetCommInfoStatus()) {
196 0 : KfcCommand kfcCmd = KfcCommand::kNone;
197 0 : (void)hcclAicpu->BackGroundGetCmd(kfcCmd);
198 0 : if (kfcCmd == KfcCommand::kSwitchNic) {
199 0 : auto groupName = hcclAicpu->GetGroupName();
200 0 : HCCL_RUN_INFO(
201 : "[ExecutorTracer][%s]group name[%s], aicpu start switch nic", __func__, groupName.c_str());
202 0 : HcclResult ret = hcclAicpu->SwitchNic();
203 :
204 0 : KfcExecStatus switchResp;
205 0 : if (ret == HCCL_SUCCESS) {
206 0 : switchResp.execStatus.kfcStatus = KfcStatus::kSwitchSuccess;
207 0 : (void)hcclAicpu->ResponseBackGroundStatus(switchResp);
208 : } else {
209 0 : switchResp.execStatus.kfcStatus = KfcStatus::kSwitchFail;
210 0 : (void)hcclAicpu->ResponseBackGroundStatus(switchResp);
211 : }
212 0 : HCCL_INFO(
213 : "[ExecutorTracer][%s]group name[%s], aicpu finish switch nic, ret[%d]", __func__, groupName.c_str(),
214 : ret);
215 0 : }
216 : }
217 : }
218 8100 : }
219 :
220 8100 : void ExecutorTracer::HandleResumeChangeLink([[maybe_unused]] AicpuComContext* const ctx)
221 : {
222 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
223 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
224 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
225 8100 : for (auto& commItem : aicpuCommInfo) {
226 0 : hccl::HcclCommAicpu* hcclAicpu = commItem.second;
227 0 : if (hcclAicpu == nullptr) {
228 0 : HCCL_ERROR("[ExecutorTracer][%s]hcclAicpu is nullptr", __func__);
229 : }
230 0 : if (hcclAicpu != nullptr && hcclAicpu->GetCommInfoStatus()) {
231 0 : KfcCommand kfcCmd = KfcCommand::kNone;
232 0 : (void)hcclAicpu->BackGroundGetCmd(kfcCmd);
233 0 : if (kfcCmd == KfcCommand::NsChangeLink) {
234 0 : auto groupName = hcclAicpu->GetGroupName();
235 0 : HCCL_INFO(
236 : "[ExecutorTracer][resume][%s]group name[%s], resume aicpu, start change link", __func__,
237 : groupName.c_str());
238 0 : HcclResult ret = hcclAicpu->ResumeChangeLink();
239 0 : KfcExecStatus resumeResp;
240 0 : if (ret == HCCL_SUCCESS) {
241 0 : resumeResp.execStatus.kfcStatus = KfcStatus::kResumeChanged;
242 0 : HCCL_INFO(
243 : "[ExecutorTracer][resume][%s]group name[%s], resume aicpu, change link, kResumeChanged",
244 : __func__, groupName.c_str());
245 : } else {
246 0 : resumeResp.execStatus.kfcStatus = KfcStatus::kResumeError;
247 0 : HCCL_INFO(
248 : "[ExecutorTracer][resume][%s]group name[%s], resume aicpu, change link, kResumeError", __func__,
249 : groupName.c_str());
250 : }
251 0 : (void)hcclAicpu->ResponseBackGroundStatus(resumeResp);
252 0 : HCCL_INFO(
253 : "[ExecutorTracer][%s]group name[%s], resume process, finish change link, ret[%d]", __func__,
254 : groupName.c_str(), ret);
255 0 : }
256 : }
257 : }
258 8100 : }
259 :
260 8100 : void ExecutorTracer::HandleCqeStatusInComm()
261 : {
262 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
263 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
264 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
265 :
266 8100 : for (auto& commInfo : aicpuCommInfo) {
267 0 : std::vector<hccl::Stream> streams;
268 0 : hccl::HcclCommAicpu* hcclAicpu = commInfo.second;
269 :
270 : // 通信域走自定义算子流程初始化，需要轮询thread状态
271 0 : if (hcclAicpu->GetIsInitIndOp()) {
272 0 : hcclAicpu->HandleIndOpCqe();
273 : }
274 :
275 0 : if (!hcclAicpu->GetCommInfoStatus()) { // 已结束, 不再轮询
276 0 : continue;
277 : }
278 0 : DfxExtendInfo* dfxInfo = hcclAicpu->GetDfxExtendInfo();
279 0 : if ((dfxInfo->cqeStatus != dfx::CqeStatus::kDefault) && (dfxInfo->cqeStatus != dfx::CqeStatus::kCqeException)) {
280 0 : continue;
281 : }
282 :
283 0 : (void)hcclAicpu->GetStreamAll(streams);
284 0 : for (hccl::Stream& stream : streams) {
285 0 : hcclAicpu->HandleCqeException(stream, false);
286 : }
287 0 : }
288 8100 : }
289 :
290 8100 : void ExecutorTracer::HandleReportStatusInComm()
291 : {
292 8100 : std::shared_lock<std::shared_mutex> rwlock(AicpuHcclProcess::AicpuGetCommMutex());
293 8100 : std::vector<std::pair<std::string, hccl::HcclCommAicpu*>> aicpuCommInfo;
294 8100 : (void)AicpuHcclProcess::AicpuGetCommAll(aicpuCommInfo);
295 :
296 8100 : for (auto& commInfo : aicpuCommInfo) {
297 0 : hccl::HcclCommAicpu* hcclAicpu = commInfo.second;
298 :
299 0 : if (!hcclAicpu || !hcclAicpu->GetCommInfoStatus()) { // 已结束, 不再轮询
300 0 : continue;
301 : }
302 :
303 0 : u32 deviceId = hcclAicpu->GetDevId();
304 :
305 0 : std::queue<dfx::ReportStatus> reportStatusQueue;
306 0 : (void)hcclAicpu->GetReportStatusQueue(reportStatusQueue);
307 :
308 0 : while (!reportStatusQueue.empty()) {
309 0 : dfx::ReportStatus reportStatus = reportStatusQueue.front();
310 0 : HCCL_INFO(
311 : "Reporting opRetry status[%d] to dp frame, deviceId[%u], report queue size[%zu].",
312 : static_cast<int>(reportStatus), deviceId, reportStatusQueue.size());
313 0 : HcclResult ret = dfx::CannErrorReporter::GetInstance().UpdateSensorNode(deviceId, reportStatus);
314 0 : if (ret != HCCL_SUCCESS) {
315 0 : HCCL_WARNING(
316 : "Fail to report reportStatus[%d] to dp frame, status dropped, deviceId[%u].",
317 : static_cast<int>(reportStatus), deviceId);
318 : }
319 0 : reportStatusQueue.pop();
320 : }
321 0 : }
322 8100 : }
323 :
324 8100 : void ExecutorTracer::HandleCqeStatus(AicpuComContext* const ctx)
325 : {
326 8100 : HandleCqeStatusInComm();
327 8100 : AicpuExecutorTracer::HandleCqeStatus(ctx);
328 8100 : }
329 :
330 0 : void ExecutorTracer::SetCqeQueryInput(
331 : const uint32_t devId, const HcclComStreamInfo& streamInfo, CqeQueryInput& cqeQueryInput)
332 : {
333 0 : cqeQueryInput.devId = devId;
334 0 : cqeQueryInput.streamId = streamInfo.actualStreamId;
335 0 : cqeQueryInput.sqId = streamInfo.sqId;
336 0 : cqeQueryInput.cqId = streamInfo.logicCqId;
337 0 : cqeQueryInput.type = static_cast<uint32_t>(DRV_LOGIC_TYPE);
338 0 : }
339 :
340 0 : void ExecutorTracer::HandleAICPUCommand(hccl::HcclCommAicpu* const commInfo)
341 : {
342 : using CommandCall = std::function<void(hccl::HcclCommAicpu* const commInfo)>;
343 : static std::map<KfcCommand, CommandCall> commandAicpuHandles
344 0 : = {{KfcCommand::NsStopExec, AICPUcommandHandles::NsCommStop},
345 0 : {KfcCommand::NsClear, AICPUcommandHandles::NsCommClean}};
346 0 : KfcCommand cmd = KfcCommand::kNone;
347 0 : (void)commInfo->BackGroundGetCmd(cmd);
348 0 : auto iter = commandAicpuHandles.find(cmd);
349 0 : if (iter == commandAicpuHandles.cend()) {
350 0 : return;
351 : }
352 0 : HCCL_RUN_INFO("Group[%s] start to run aicpu command %d", commInfo->GetGroupName().c_str(), static_cast<int>(cmd));
353 0 : iter->second(commInfo);
354 0 : }
355 :
356 0 : void AICPUcommandHandles::NsCommStop(hccl::HcclCommAicpu* const commInfo)
357 : {
358 0 : bool streamStatus = commInfo->GetCommInfoStreamStatus();
359 0 : if (streamStatus) {
360 0 : std::string groupName = commInfo->GetGroupName();
361 0 : commInfo->SetCommInfoStreamStatus(false);
362 0 : HCCL_RUN_INFO("[NsRecovery][NsCommStop] groupName[%s]", groupName.c_str());
363 0 : commInfo->NsCommStop();
364 0 : }
365 0 : }
366 :
367 0 : void AICPUcommandHandles::NsCommClean(hccl::HcclCommAicpu* const commInfo)
368 : {
369 0 : bool streamStatus = commInfo->GetCommInfoStreamStatus();
370 0 : if (!streamStatus) {
371 0 : std::string groupName = commInfo->GetGroupName();
372 0 : commInfo->SetCommInfoStreamStatus(true);
373 0 : HCCL_RUN_INFO("[NsRecovery][NsCommClean] groupName[%s]", groupName.c_str());
374 0 : commInfo->NsCommClean();
375 0 : commInfo->SetCommRecoveryFlag(false);
376 0 : }
377 0 : }
378 : } // namespace dfx_tracer
|
