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 <sstream>
12 : #include <iostream>
13 : #include <cstdint>
14 : #include <iomanip>
15 : #include <array>
16 : #include "adapter_rts_common.h"
17 : #include "externalinput_pub.h"
18 : #include "sal_pub.h"
19 : #include "../../../algorithm/pub_inc/common.h"
20 : #include "acl/error_codes/rt_error_codes.h"
21 : #include "task_exception_handler.h"
22 :
23 : using namespace hccl;
24 : using namespace std;
25 : std::atomic<int> TaskExceptionHandler::communicatorCount_{0};
26 : std::atomic<bool> TaskExceptionHandler::errMsgFlag_{false};
27 : GetErrStatusVecCallBack g_GetErrStatusVecCallBack = nullptr;
28 : std::mutex g_communicatorCallbackMapMutex;
29 : array<map<s32, GetAicpuTaskExceptionCallBack>, MAX_MODULE_DEVICE_NUM> g_communicatorCallbackMap;
30 : std::mutex g_commHadCallbackArrayMutex;
31 : array<bool, MAX_MODULE_DEVICE_NUM> g_commHadCallbackArray = {false};
32 : #ifdef __cplusplus
33 : extern "C" {
34 : #endif // __cplusplus
35 46 : void RegisterGetErrStatusVecCallBack(GetErrStatusVecCallBack p1)
36 : {
37 46 : g_GetErrStatusVecCallBack = p1;
38 46 : return;
39 : }
40 :
41 18 : void RegisterGetAicpuTaskExceptionCallBack(s32 streamId, u32 deviceLogicId, GetAicpuTaskExceptionCallBack p1)
42 : {
43 18 : if (deviceLogicId >= MAX_MODULE_DEVICE_NUM) {
44 0 : HCCL_ERROR(
45 : "[RegisterGetAicpuTaskExceptionCallBack] deviceLogicId[%u] out of range, max is %u", deviceLogicId,
46 : MAX_MODULE_DEVICE_NUM - 1);
47 0 : return;
48 : }
49 18 : lock_guard<mutex> lock(g_communicatorCallbackMapMutex);
50 18 : g_communicatorCallbackMap[deviceLogicId][streamId] = p1;
51 18 : return;
52 18 : }
53 :
54 3 : void UnregisterGetAicpuTaskExceptionCallBack(s32 streamId, u32 deviceLogicId)
55 : {
56 3 : if (deviceLogicId >= MAX_MODULE_DEVICE_NUM) {
57 0 : HCCL_ERROR(
58 : "[UnregisterGetAicpuTaskExceptionCallBack] deviceLogicId[%u] out of range, max is %u", deviceLogicId,
59 : MAX_MODULE_DEVICE_NUM - 1);
60 0 : return;
61 : }
62 3 : lock_guard<mutex> lock(g_communicatorCallbackMapMutex);
63 3 : auto& deviceMap = g_communicatorCallbackMap[deviceLogicId];
64 3 : auto it = deviceMap.find(streamId);
65 3 : if (it != deviceMap.end()) {
66 2 : deviceMap.erase(it);
67 : }
68 3 : return;
69 3 : }
70 : #ifdef __cplusplus
71 : }
72 : #endif // __cplusplus
73 : namespace hccl {
74 : namespace hccl_alg {
75 17 : std::vector<std::string> GetErrStatusVec(s32 deviceLogicID, const std::string& group = HCCL_WORLD_GROUP)
76 : {
77 17 : if (g_GetErrStatusVecCallBack != nullptr) {
78 17 : return g_GetErrStatusVecCallBack(deviceLogicID, group);
79 : } else {
80 0 : HCCL_RUN_WARNING("[GetErrStatusVec]g_GetErrStatusVecCallBack is nullptr.");
81 : }
82 0 : return std::vector<std::string>();
83 : }
84 : } // namespace hccl_alg
85 : } // namespace hccl
86 :
87 : std::string GetTaskName(TaskType taskType, bool isAlgInfo = false);
88 : std::string GetLinkTypeName(LinkType linkInput);
89 : std::string GetAlgTypeStr(AlgType algType);
90 : std::string GetTaskBriefsName(TaskType taskType);
91 :
92 : namespace {
93 : constexpr u32 STREAM_COUNT_UPPER_LIMIT = 2048; // stream 数量最大值2048，防止内存占用量过大
94 : constexpr u32 TASK_COUNT_UPPER_LIMIT = 2048; // task 数量最大值2048，防止内存占用量过大
95 : constexpr u32 TASK_COUNT_UPPER_LIMIT_OP_BASE = 65535; // 单算子模式task数量最大值
96 : constexpr u32 TASK_CONTEXT_SIZE = 50; // task 执行失败时打印前序task的数量
97 : constexpr u32 TASK_CONTEXT_INFO_SIZE = LOG_TMPBUF_SIZE - 50; // task 执行失败时打印前序task信息的长度限制
98 : constexpr u32 PRINT_TASK_AIV_INFO_COUNT = 10;
99 : constexpr u32 AIV_KERNEL_FLAG_SIZE_PER_OP = 6;
100 :
101 : constexpr u32 MAX_NUM_BLOCKS = 48;
102 : constexpr u32 MAX_RANK_SIZE_SUPERPOD = 768;
103 : constexpr u32 INTERVAL_1VN = 128;
104 : constexpr u32 INTERVAL_NV1 = 128;
105 : constexpr u32 INTERVAL_1V1 = 8;
106 : constexpr u32 PING_PONG_NUM = 2;
107 : constexpr u32 PRINT_NV1_NUM = 4;
108 : constexpr u32 PRINT_1VN_NUM = 4;
109 : constexpr u32 INTERVAL_COUNT = 8;
110 : constexpr u32 NOTIFY_NUM = 3;
111 : constexpr u32 NUM_BLOCKS_PER_RANK = 4;
112 : constexpr u32 CORE_PER_CARDS = 4;
113 : constexpr u32 NOTIFY_GROUPS_1V1 = 2;
114 :
115 : u32 maxStrCount = 0;
116 : u32 maxTaskCount = 0;
117 :
118 2 : std::string GetReduceOpString(HcclReduceOp op)
119 : {
120 2 : u32 opVal = static_cast<u32>(op);
121 2 : return opVal < ProfilerBase::opString.size() ? std::to_string(ProfilerBase::opString[opVal]) :
122 4 : "Unknown(" + std::to_string(op) + ")";
123 : }
124 :
125 2 : std::string GetDataTypeString(HcclDataType dataType)
126 : {
127 2 : u32 dtVal = static_cast<u32>(dataType);
128 2 : return dtVal < ProfilerBase::dataTypeString.size() ? std::to_string(ProfilerBase::dataTypeString[dtVal]) :
129 4 : "Unknown(" + std::to_string(dataType) + ")";
130 : }
131 : } // namespace
132 : array<map<int, shared_ptr<deque<TaskInfo>>>, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::taskMap;
133 : array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::taskMapMutex;
134 : array<map<int, shared_ptr<deque<FFTSOpInfo>>>, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::opMap;
135 : array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::opMapMutex;
136 : array<
137 : std::map<
138 : int, shared_ptr<std::deque<std::pair<std::shared_ptr<FFTSOpInfo>, std::shared_ptr<std::vector<CtxInfo>>>>>>,
139 : MAX_MODULE_DEVICE_NUM>
140 : TaskExceptionHandler::opCtxInfo;
141 : array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::opCtxInfoMutex;
142 : array<std::vector<CtxInfo>, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::ctxInfoArray;
143 : array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::ctxInfoVectorMutex;
144 : array<std::map<const std::string, std::pair<const std::string, std::shared_ptr<GroupRankInfo>>>, MAX_MODULE_DEVICE_NUM>
145 : TaskExceptionHandler::groupRankMap;
146 : array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::groupRankMapMutex;
147 : array<std::map<const std::string, std::shared_ptr<std::queue<OpDataInfo>>>, MAX_MODULE_DEVICE_NUM>
148 : TaskExceptionHandler::tagOpDataMap;
149 : array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::tagOpDataMapMutex;
150 : std::array<std::map<const std::string, std::string>, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::groupUdiMap;
151 : std::array<std::mutex, MAX_MODULE_DEVICE_NUM> TaskExceptionHandler::groupUdiMapMutex;
152 34 : TaskInfo::TaskInfo(
153 34 : u32& streamID, u32& taskID, string& tag, TaskType& taskType, AlgType& algType, u32& index, const TaskParaDMA& para)
154 34 : : streamID(streamID),
155 34 : taskID(taskID),
156 34 : tag(tag),
157 34 : taskType(taskType),
158 34 : isAlgInfo(false),
159 34 : algType(algType),
160 34 : index(index)
161 : {
162 34 : taskPara.DMA.src = para.src;
163 34 : taskPara.DMA.dst = para.dst;
164 34 : taskPara.DMA.size = para.size;
165 34 : taskPara.DMA.notifyID = para.notifyID;
166 34 : taskPara.DMA.linkType = para.linkType;
167 34 : taskPara.DMA.remoteUserRank = para.remoteUserRank;
168 34 : }
169 10 : TaskInfo::TaskInfo(
170 : u32& streamID, u32& taskID, string& tag, TaskType& taskType, AlgType& algType, u32& index,
171 10 : const TaskParaReduce& para)
172 10 : : streamID(streamID),
173 10 : taskID(taskID),
174 10 : tag(tag),
175 10 : taskType(taskType),
176 10 : isAlgInfo(false),
177 10 : algType(algType),
178 10 : index(index)
179 : {
180 10 : taskPara.Reduce.src = para.src;
181 10 : taskPara.Reduce.dst = para.dst;
182 10 : taskPara.Reduce.size = para.size;
183 10 : taskPara.Reduce.op = para.op;
184 10 : taskPara.Reduce.dataType = para.dataType;
185 10 : taskPara.Reduce.linkType = para.linkType;
186 10 : taskPara.Reduce.remoteUserRank = para.remoteUserRank;
187 10 : }
188 12 : TaskInfo::TaskInfo(
189 : u32& streamID, u32& taskID, string& tag, TaskType& taskType, AlgType& algType, u32& index,
190 12 : const TaskParaNotify& para)
191 12 : : streamID(streamID),
192 12 : taskID(taskID),
193 12 : tag(tag),
194 12 : taskType(taskType),
195 12 : isAlgInfo(false),
196 12 : algType(algType),
197 12 : index(index)
198 : {
199 12 : taskPara.Notify.notifyID = para.notifyID;
200 12 : taskPara.Notify.stage = para.stage;
201 12 : taskPara.Notify.remoteUserRank = para.remoteUserRank;
202 12 : }
203 5 : TaskInfo::TaskInfo(u32& streamID, u32& taskID, string& tag, const TaskParaAiv& para)
204 5 : : streamID(streamID),
205 5 : taskID(taskID),
206 5 : tag(tag),
207 5 : isAlgInfo(true)
208 : {
209 5 : taskPara.Aiv.cmdType = para.cmdType;
210 5 : taskPara.Aiv.tag = para.tag;
211 5 : taskPara.Aiv.size = para.size;
212 5 : taskPara.Aiv.numBlocks = para.numBlocks;
213 5 : taskPara.Aiv.rankSize = para.rankSize;
214 5 : taskPara.Aiv.flagMem = para.flagMem;
215 5 : taskPara.Aiv.aivRdmaStep = para.aivRdmaStep;
216 5 : taskPara.Aiv.rank = para.rank;
217 5 : taskPara.Aiv.isOpbase = para.isOpbase;
218 5 : }
219 7 : CtxInfo::CtxInfo(TaskType& taskType, const TaskParaDMA& para) : taskType(taskType)
220 : {
221 7 : ctxPara.DMA.src = para.src;
222 7 : ctxPara.DMA.dst = para.dst;
223 7 : ctxPara.DMA.size = para.size;
224 7 : ctxPara.DMA.notifyID = para.notifyID;
225 7 : ctxPara.DMA.linkType = para.linkType;
226 7 : ctxPara.DMA.remoteUserRank = para.remoteUserRank;
227 7 : }
228 0 : CtxInfo::CtxInfo(TaskType& taskType, const TaskParaReduce& para) : taskType(taskType)
229 : {
230 0 : ctxPara.Reduce.src = para.src;
231 0 : ctxPara.Reduce.dst = para.dst;
232 0 : ctxPara.Reduce.size = para.size;
233 0 : ctxPara.Reduce.op = para.op;
234 0 : ctxPara.Reduce.dataType = para.dataType;
235 0 : ctxPara.Reduce.linkType = para.linkType;
236 0 : ctxPara.Reduce.remoteUserRank = para.remoteUserRank;
237 0 : }
238 2 : CtxInfo::CtxInfo(TaskType& taskType, const TaskParaNotify& para) : taskType(taskType)
239 : {
240 2 : ctxPara.Notify.notifyID = para.notifyID;
241 2 : ctxPara.Notify.stage = para.stage;
242 2 : ctxPara.Notify.remoteUserRank = para.remoteUserRank;
243 2 : }
244 :
245 21 : string TaskInfo::GetBaseInfoStr() // 防止tag字符串过长，base信息和para信息分开打印
246 : {
247 21 : string taskContent;
248 21 : taskContent += "streamID:[";
249 21 : taskContent += std::to_string(streamID);
250 21 : taskContent += "], taskID[";
251 21 : taskContent += std::to_string(taskID);
252 21 : taskContent += "], taskType[";
253 21 : taskContent += GetTaskName(taskType, isAlgInfo);
254 21 : taskContent += "], tag[";
255 21 : taskContent += tag;
256 21 : taskContent += "], ";
257 21 : taskContent += GetAlgTypeStr(algType);
258 21 : return taskContent;
259 0 : }
260 :
261 0 : string TaskInfo::GetRankInfo()
262 : {
263 0 : u32 remoteRank = INVALID_VALUE_RANKID;
264 0 : switch (taskType) {
265 0 : case TaskType::TASK_SDMA:
266 : case TaskType::TASK_RDMA:
267 0 : remoteRank = taskPara.DMA.remoteUserRank;
268 0 : break;
269 0 : case TaskType::TASK_REDUCE_INLINE:
270 : case TaskType::TASK_REDUCE_TBE:
271 0 : remoteRank = taskPara.Reduce.remoteUserRank;
272 0 : break;
273 0 : case TaskType::TASK_NOTIFY_RECORD:
274 : case TaskType::TASK_NOTIFY_WAIT:
275 0 : remoteRank = taskPara.Notify.remoteUserRank;
276 0 : break;
277 0 : default:
278 0 : return "/";
279 : }
280 0 : return (remoteRank == INVALID_VALUE_RANKID) ? "/" : to_string(remoteRank);
281 : }
282 :
283 0 : string TaskInfo::GetNotifyInfo()
284 : {
285 0 : u64 notifyInfo = INVALID_U64;
286 0 : switch (taskType) {
287 0 : case TaskType::TASK_RDMA:
288 0 : notifyInfo = taskPara.DMA.notifyID;
289 0 : break;
290 0 : case TaskType::TASK_NOTIFY_RECORD:
291 : case TaskType::TASK_NOTIFY_WAIT:
292 0 : notifyInfo = taskPara.Notify.notifyID;
293 0 : break;
294 0 : default:
295 0 : return "/";
296 : }
297 0 : if (notifyInfo == INVALID_U64) {
298 0 : return "/";
299 : } else {
300 0 : stringstream paraStr;
301 : // NotifyId取后八位16进制数进行打印
302 0 : paraStr << std::hex << static_cast<u32>(notifyInfo);
303 0 : return paraStr.str();
304 0 : }
305 : }
306 :
307 18 : string TaskInfo::GetParaInfoStr()
308 : {
309 18 : if (isAlgInfo) {
310 0 : return GetParaAiv();
311 : }
312 18 : switch (taskType) {
313 5 : case TaskType::TASK_SDMA:
314 : case TaskType::TASK_RDMA:
315 5 : return GetParaDMA();
316 2 : case TaskType::TASK_REDUCE_INLINE:
317 : case TaskType::TASK_REDUCE_TBE:
318 2 : return GetParaReduce();
319 11 : case TaskType::TASK_NOTIFY_RECORD:
320 : case TaskType::TASK_NOTIFY_WAIT:
321 11 : return GetParaNotify();
322 0 : default:
323 0 : return "unknown task";
324 : }
325 : }
326 :
327 5 : string TaskInfo::GetParaDMA()
328 : {
329 5 : string retStr;
330 5 : stringstream paraStr;
331 5 : paraStr << "src:" << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(taskPara.DMA.src))
332 : << "], dst:"
333 5 : << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(taskPara.DMA.dst)) << "], size:"
334 5 : << "[0x" << std::hex << static_cast<u64>(taskPara.DMA.size) << "], notify id:"
335 5 : << "[0x" << std::hex << std::setw(16) // 16字符长度对齐
336 5 : << std::setfill('0') << taskPara.DMA.notifyID << "], link type:[" << GetLinkTypeName(taskPara.DMA.linkType)
337 : << "], remote rank:["
338 10 : << ((taskPara.DMA.remoteUserRank == INVALID_VALUE_RANKID) ? "local" :
339 7 : to_string(taskPara.DMA.remoteUserRank))
340 10 : << "]";
341 5 : retStr += paraStr.str();
342 5 : return retStr;
343 5 : }
344 :
345 11 : string TaskInfo::GetParaNotify()
346 : {
347 11 : string retStr;
348 11 : stringstream paraStr;
349 : paraStr << "notify id:"
350 11 : << "[0x" << std::hex << std::setw(16) // 16字节长度对齐
351 11 : << std::setfill('0') << taskPara.Notify.notifyID << "], stage:[" << taskPara.Notify.stage
352 : << "], remote rank:["
353 22 : << ((taskPara.Notify.remoteUserRank == INVALID_VALUE_RANKID) ? "local" :
354 11 : to_string(taskPara.Notify.remoteUserRank))
355 22 : << "]";
356 11 : retStr += paraStr.str();
357 11 : return retStr;
358 11 : }
359 :
360 2 : string TaskInfo::GetParaReduce()
361 : {
362 2 : string retStr;
363 2 : stringstream paraStr;
364 2 : paraStr << "src:" << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(taskPara.Reduce.src))
365 : << "], dst:"
366 2 : << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(taskPara.Reduce.dst)) << "], size:"
367 2 : << "[0x" << std::hex
368 2 : << static_cast<u64>(taskPara.Reduce.size * ProfilerBase::sizeOf[taskPara.Reduce.dataType]) << "], op:["
369 2 : << GetReduceOpString(taskPara.Reduce.op) << "], data type:[" << GetDataTypeString(taskPara.Reduce.dataType)
370 2 : << "], link type:[" << GetLinkTypeName(taskPara.Reduce.linkType) << "], remote rank:["
371 4 : << ((taskPara.Reduce.remoteUserRank == INVALID_VALUE_RANKID) ? "local" :
372 4 : to_string(taskPara.Reduce.remoteUserRank))
373 4 : << "]";
374 2 : retStr += paraStr.str();
375 2 : return retStr;
376 2 : }
377 :
378 0 : string TaskInfo::GetParaAiv()
379 : {
380 0 : string retStr;
381 0 : stringstream paraStr;
382 0 : paraStr << "cmdType:[" << static_cast<int>(taskPara.Aiv.cmdType) << "], "
383 0 : << "tag:[" << taskPara.Aiv.tag << "], "
384 0 : << "size:[" << taskPara.Aiv.size << "], "
385 0 : << "numBlocks:[" << taskPara.Aiv.numBlocks << "], "
386 0 : << "rankSize:[" << taskPara.Aiv.rankSize << "], "
387 0 : << "aivRdmaStep:[" << taskPara.Aiv.aivRdmaStep << "], "
388 0 : << "flagMem:[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(taskPara.Aiv.flagMem)) << "], "
389 0 : << "isOpbase:[" << taskPara.Aiv.isOpbase << "]";
390 :
391 0 : retStr += paraStr.str();
392 0 : return retStr;
393 0 : }
394 :
395 7 : u32 TaskInfo::GetRemoteUserRank() { return taskPara.Notify.remoteUserRank; }
396 :
397 6 : string CtxInfo::GetCtxBaseInfoStr() // 防止tag字符串过长，base信息和para信息分开打印
398 : {
399 6 : string taskContent;
400 6 : taskContent += "taskType[";
401 6 : taskContent += GetTaskName(taskType);
402 6 : taskContent += "].";
403 6 : return taskContent;
404 0 : }
405 :
406 0 : string CtxInfo::GetCtxRankInfo()
407 : {
408 0 : u32 remoteRank = INVALID_VALUE_RANKID;
409 0 : switch (taskType) {
410 0 : case TaskType::TASK_SDMA:
411 : case TaskType::TASK_RDMA:
412 0 : remoteRank = ctxPara.DMA.remoteUserRank;
413 0 : break;
414 0 : case TaskType::TASK_REDUCE_INLINE:
415 : case TaskType::TASK_REDUCE_TBE:
416 0 : remoteRank = ctxPara.Reduce.remoteUserRank;
417 0 : break;
418 0 : case TaskType::TASK_NOTIFY_RECORD:
419 : case TaskType::TASK_NOTIFY_WAIT:
420 0 : remoteRank = ctxPara.Notify.remoteUserRank;
421 0 : break;
422 0 : default:
423 0 : return "/";
424 : }
425 0 : return (remoteRank == INVALID_VALUE_RANKID) ? "/" : to_string(remoteRank);
426 : }
427 :
428 0 : string CtxInfo::GetCtxNotifyInfo()
429 : {
430 0 : u64 notifyInfo = INVALID_U64;
431 0 : switch (taskType) {
432 0 : case TaskType::TASK_RDMA:
433 0 : notifyInfo = ctxPara.DMA.notifyID;
434 0 : break;
435 0 : case TaskType::TASK_NOTIFY_RECORD:
436 : case TaskType::TASK_NOTIFY_WAIT:
437 0 : notifyInfo = ctxPara.Notify.notifyID;
438 0 : break;
439 0 : default:
440 0 : return "/";
441 : }
442 0 : if (notifyInfo == INVALID_U64) {
443 0 : return "/";
444 : } else {
445 0 : stringstream paraStr;
446 : // NotifyId取后八位16进制数进行打印
447 0 : paraStr << std::hex << static_cast<u32>(notifyInfo);
448 0 : return paraStr.str();
449 0 : }
450 : }
451 :
452 4 : string CtxInfo::GetCtxParaInfoStr()
453 : {
454 4 : switch (taskType) {
455 1 : case TaskType::TASK_SDMA:
456 : case TaskType::TASK_RDMA:
457 1 : return GetCtxParaDMA();
458 0 : case TaskType::TASK_REDUCE_INLINE:
459 : case TaskType::TASK_REDUCE_TBE:
460 0 : return GetCtxParaReduce();
461 3 : case TaskType::TASK_NOTIFY_RECORD:
462 : case TaskType::TASK_NOTIFY_WAIT:
463 3 : return GetCtxParaNotify();
464 0 : default:
465 0 : return "unknown task";
466 : }
467 : }
468 :
469 1 : string CtxInfo::GetCtxParaDMA()
470 : {
471 1 : string retStr;
472 1 : stringstream paraStr;
473 1 : paraStr << "src:" << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(ctxPara.DMA.src))
474 : << "], dst:"
475 1 : << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(ctxPara.DMA.dst)) << "], size:"
476 1 : << "[0x" << std::hex << static_cast<u64>(ctxPara.DMA.size) << "], notify id:"
477 1 : << "[0x" << std::hex << std::setw(16) // 16字符长度对齐
478 1 : << std::setfill('0') << ctxPara.DMA.notifyID << "], link type:[" << GetLinkTypeName(ctxPara.DMA.linkType)
479 : << "], remote rank:["
480 3 : << ((ctxPara.DMA.remoteUserRank == INVALID_VALUE_RANKID) ? "local" : to_string(ctxPara.DMA.remoteUserRank))
481 2 : << "]";
482 1 : retStr += paraStr.str();
483 1 : return retStr;
484 1 : }
485 :
486 5 : string CtxInfo::GetCtxParaNotify()
487 : {
488 5 : string retStr;
489 5 : stringstream paraStr;
490 : paraStr << "notify id:"
491 5 : << "[0x" << std::hex << std::setw(16) // 16字节长度对齐
492 5 : << std::setfill('0') << ctxPara.Notify.notifyID << "], stage:[" << ctxPara.Notify.stage
493 : << "], remote rank:["
494 10 : << ((ctxPara.Notify.remoteUserRank == INVALID_VALUE_RANKID) ? "local" :
495 5 : to_string(ctxPara.Notify.remoteUserRank))
496 10 : << "]";
497 5 : retStr += paraStr.str();
498 5 : return retStr;
499 5 : }
500 :
501 0 : string CtxInfo::GetCtxParaReduce()
502 : {
503 0 : string retStr;
504 0 : stringstream paraStr;
505 0 : paraStr << "src:" << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(ctxPara.Reduce.src))
506 : << "], dst:"
507 0 : << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(ctxPara.Reduce.dst)) << "], size:"
508 0 : << "[0x" << std::hex
509 0 : << static_cast<u64>(ctxPara.Reduce.size * ProfilerBase::sizeOf[ctxPara.Reduce.dataType]) << "], op:["
510 0 : << GetReduceOpString(ctxPara.Reduce.op) << "], data type:[" << GetDataTypeString(ctxPara.Reduce.dataType)
511 0 : << "], link type:[" << GetLinkTypeName(ctxPara.Reduce.linkType) << "], remote rank:["
512 0 : << ((ctxPara.Reduce.remoteUserRank == INVALID_VALUE_RANKID) ? "local" :
513 0 : to_string(ctxPara.Reduce.remoteUserRank))
514 0 : << "]";
515 0 : retStr += paraStr.str();
516 0 : return retStr;
517 0 : }
518 :
519 1 : u32 CtxInfo::GetCtxRemoteUserRank() { return ctxPara.Notify.remoteUserRank; }
520 :
521 27 : std::string GetTaskName(TaskType taskType, bool isAlgInfo)
522 : {
523 27 : std::string taskName;
524 :
525 27 : if (isAlgInfo) {
526 0 : taskName = "Task AIV";
527 0 : return taskName;
528 : }
529 :
530 27 : switch (taskType) {
531 5 : case TaskType::TASK_SDMA:
532 5 : taskName += "Memcpy";
533 5 : break;
534 6 : case TaskType::TASK_RDMA:
535 6 : taskName += "RDMASend";
536 6 : break;
537 2 : case TaskType::TASK_REDUCE_INLINE:
538 2 : taskName += "Reduce Inline";
539 2 : break;
540 0 : case TaskType::TASK_REDUCE_TBE:
541 0 : taskName += "Reduce TBE";
542 0 : break;
543 0 : case TaskType::TASK_NOTIFY_RECORD:
544 0 : taskName += "Notify Record";
545 0 : break;
546 14 : case TaskType::TASK_NOTIFY_WAIT:
547 14 : taskName += "Notify Wait";
548 14 : break;
549 0 : default:
550 0 : return "unknown task";
551 : }
552 :
553 27 : return taskName;
554 27 : }
555 0 : std::string GetTaskBriefsName(TaskType taskType)
556 : {
557 0 : std::string taskName;
558 0 : switch (taskType) {
559 0 : case TaskType::TASK_SDMA:
560 0 : taskName += "M";
561 0 : break;
562 0 : case TaskType::TASK_RDMA:
563 0 : taskName += "RS";
564 0 : break;
565 0 : case TaskType::TASK_REDUCE_INLINE:
566 0 : taskName += "IR";
567 0 : break;
568 0 : case TaskType::TASK_REDUCE_TBE:
569 0 : taskName += "R";
570 0 : break;
571 0 : case TaskType::TASK_NOTIFY_RECORD:
572 0 : taskName += "NR";
573 0 : break;
574 0 : case TaskType::TASK_NOTIFY_WAIT:
575 0 : taskName += "NW";
576 0 : break;
577 0 : default:
578 0 : return "unknown task";
579 : }
580 :
581 0 : return taskName;
582 0 : }
583 8 : std::string GetLinkTypeName(LinkType linkInput)
584 : {
585 8 : switch (linkInput) {
586 8 : case LinkType::LINK_ONCHIP:
587 16 : return "OnChip";
588 0 : case LinkType::LINK_HCCS:
589 0 : return "HCCS";
590 0 : case LinkType::LINK_PCIE:
591 0 : return "PCIe";
592 0 : case LinkType::LINK_ROCE:
593 0 : return "RoCE";
594 0 : case LinkType::LINK_SIO:
595 0 : return "SIO";
596 0 : case LinkType::LINK_HCCS_SW:
597 0 : return "HCCS_SW";
598 0 : default:
599 0 : return "OnChip";
600 : }
601 : }
602 :
603 28 : std::string GetAlgTypeStr(AlgType algType)
604 : {
605 28 : std::string algTypeStr = "";
606 28 : algTypeStr += "AlgType(level 0-1-2):[";
607 28 : auto alg0It = HCCL_ALGO_LEVEL0_NAME_MAP.find(algType.algoLevel0);
608 28 : if (alg0It != HCCL_ALGO_LEVEL0_NAME_MAP.end()) {
609 28 : algTypeStr += alg0It->second;
610 : } else {
611 0 : algTypeStr += "null";
612 : }
613 :
614 28 : algTypeStr += "-";
615 28 : auto alg1It = HCCL_ALGO_LEVEL1_NAME_MAP.find(algType.algoLevel1);
616 28 : if (alg1It != HCCL_ALGO_LEVEL1_NAME_MAP.end()) {
617 28 : algTypeStr += alg1It->second;
618 : } else {
619 0 : algTypeStr += "null";
620 : }
621 :
622 28 : algTypeStr += "-";
623 28 : auto alg2It = HCCL_ALGO_LEVEL2_NAME_MAP.find(algType.algoLevel2);
624 28 : if (alg2It != HCCL_ALGO_LEVEL2_NAME_MAP.end()) {
625 28 : algTypeStr += alg2It->second;
626 : } else {
627 0 : algTypeStr += "null";
628 : }
629 28 : algTypeStr += "].";
630 28 : return algTypeStr;
631 0 : }
632 :
633 7 : string FFTSOpInfo::GetBaseInfoStr() // 防止tag字符串过长，base信息和para信息分开打印
634 : {
635 7 : string taskContent;
636 7 : taskContent += "streamID:[";
637 7 : taskContent += std::to_string(streamID);
638 7 : taskContent += "], taskID[";
639 7 : taskContent += std::to_string(taskID);
640 7 : taskContent += "], tag[";
641 7 : taskContent += std::string(tag.get());
642 7 : taskContent += "], ";
643 7 : taskContent += GetAlgTypeStr(algType);
644 7 : return taskContent;
645 0 : }
646 332 : TaskExceptionHandler::TaskExceptionHandler(u32 deviceLogicId) : ProfilerBase(deviceLogicId) {}
647 644 : TaskExceptionHandler::~TaskExceptionHandler() {}
648 17 : std::string GetAndPrintHeartbeatErr(rtExceptionInfo* exceptionInfo, const std::string& group = HCCL_WORLD_GROUP)
649 : {
650 17 : auto errStatusVec = hccl_alg::GetErrStatusVec(exceptionInfo->deviceid, group);
651 17 : std::string errMsg = "";
652 17 : int errSize = errStatusVec.size();
653 17 : if (errSize > 0) {
654 0 : int maxListSize = 3; // 放入errMsg中的异常事件最多只有3个
655 0 : if (errSize <= maxListSize) {
656 0 : errMsg = "\nthere are(is) " + std::to_string(errSize) + " abnormal device(s):\n";
657 : } else {
658 0 : errMsg = "\nthere are " + std::to_string(errSize) + " abnormal device(s), "
659 0 : + "only the first 3 devices are listed:\n";
660 : }
661 :
662 0 : for (int i = 0; i < errSize; i++) {
663 0 : HCCL_ERROR("%s", errStatusVec[i].c_str());
664 0 : if (i < maxListSize) {
665 0 : errMsg += ("\t" + errStatusVec[i] + "\n");
666 : }
667 : }
668 : }
669 17 : return errMsg;
670 17 : }
671 2 : void TaskExceptionHandler::PrintTaskContextInfo(
672 : const std::shared_ptr<std::vector<CtxInfo>>& taskList, u32 contextId, std::string& stageErrInfo)
673 : {
674 2 : HCCL_ERROR(
675 : "%sTask run failed, context sequence before error task is "
676 : "[NotifyRecord:NR(rank,id), NotifyWait:NW(rank,id), Memcpy:M(rank), Reduce: R(rank), "
677 : "InlineReduce:IR(rank), RDMASend:RS(rank,id)]:",
678 : stageErrInfo.c_str());
679 2 : std::string taskContextInfo = "";
680 2 : u32 startIndex = (contextId > TASK_CONTEXT_SIZE) ? (contextId - TASK_CONTEXT_SIZE) : 0;
681 2 : for (; startIndex < contextId; startIndex++) {
682 0 : auto curCtxInfo = taskList->at(startIndex);
683 :
684 0 : std::string taskStr = GetTaskBriefsName(curCtxInfo.taskType);
685 0 : taskStr += "(";
686 0 : taskStr += curCtxInfo.GetCtxRankInfo();
687 0 : if (curCtxInfo.taskType == TaskType::TASK_NOTIFY_RECORD || curCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT
688 0 : || curCtxInfo.taskType == TaskType::TASK_RDMA) {
689 0 : taskStr += ("," + curCtxInfo.GetCtxNotifyInfo());
690 : }
691 0 : taskStr += "),";
692 0 : if (taskContextInfo.size() + taskStr.size() >= TASK_CONTEXT_INFO_SIZE) {
693 0 : HCCL_ERROR("%s ...", taskContextInfo.c_str());
694 0 : taskContextInfo = "";
695 : }
696 0 : taskContextInfo += taskStr;
697 0 : }
698 2 : HCCL_ERROR("%s end.", taskContextInfo.c_str());
699 4 : return;
700 2 : }
701 :
702 3 : void TaskExceptionHandler::TimeStruct2Str(struct timeval& tv, std::string& opDataContent)
703 : {
704 3 : const u32 length = 128;
705 3 : char timeStr[length] = {0};
706 3 : std::string timeStamp;
707 3 : const time_t sec = tv.tv_sec;
708 3 : struct tm nowTime = {0};
709 3 : const struct tm* tmp = localtime_r(&sec, &nowTime);
710 3 : if (tmp == nullptr) {
711 0 : return;
712 : }
713 :
714 6 : int32_t err = snprintf_s(
715 3 : timeStr, length, length - 1, "%04d-%02d-%02d-%02d:%02d:%02d.%03ld.%03ld", (nowTime.tm_year + 1900),
716 3 : nowTime.tm_mon + 1, nowTime.tm_mday, nowTime.tm_hour, nowTime.tm_min, nowTime.tm_sec, tv.tv_usec / 1000,
717 3 : tv.tv_usec % 1000);
718 3 : if (err == -1) {
719 0 : timeStamp = "unknown time";
720 : } else {
721 3 : timeStamp = timeStr;
722 : }
723 :
724 3 : opDataContent += "timeStamp:[";
725 3 : opDataContent += timeStamp;
726 3 : opDataContent += "]";
727 :
728 3 : return;
729 3 : }
730 2 : void TaskExceptionHandler::PrintOpDataInfo(OpDataInfo& opDataInfo, bool isFftsPlus, std::string& stageErrInfo)
731 : {
732 2 : stringstream opDataStr;
733 2 : opDataStr << "src" << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(opDataInfo.src))
734 2 : << "], dst[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(opDataInfo.dst)) << "], ";
735 :
736 2 : string opStr;
737 2 : if (opDataInfo.reduceType != HcclReduceOp::HCCL_REDUCE_RESERVED) {
738 0 : opStr += "reduceType[";
739 0 : opStr += GetReduceOpEnumStr(opDataInfo.reduceType);
740 0 : opStr += "], ";
741 : }
742 :
743 2 : string opDataContent;
744 2 : TimeStruct2Str(opDataInfo.tv, opDataContent);
745 2 : opDataContent += ", deviceId[";
746 2 : opDataContent += std::to_string(opDataInfo.deviceId);
747 2 : opDataContent += "], index[";
748 2 : opDataContent += std::to_string(opDataInfo.index);
749 2 : opDataContent += "], count[";
750 2 : opDataContent += std::to_string(opDataInfo.count);
751 2 : opDataContent += "], ";
752 2 : opDataContent += opStr;
753 2 : opDataContent += opDataStr.str();
754 2 : opDataContent += "dataType[";
755 2 : opDataContent += GetDataTypeEnumStr(opDataInfo.dataType);
756 2 : opDataContent += "].";
757 :
758 2 : PrintOpDataErrorLog(stageErrInfo, opDataContent);
759 4 : return;
760 2 : }
761 :
762 12 : bool TaskExceptionHandler::DealExceptionOpData(
763 : rtExceptionInfo* exceptionInfo, std::string& tag, bool isFftsPlus, u32 index, std::string& stageErrInfo)
764 : {
765 12 : bool opDataFound = false;
766 12 : std::unique_lock<std::mutex> lock(tagOpDataMapMutex[exceptionInfo->deviceid]);
767 12 : auto opDataIt = tagOpDataMap[exceptionInfo->deviceid].find(tag);
768 12 : CHK_PRT_RET(
769 : opDataIt == tagOpDataMap[exceptionInfo->deviceid].end(),
770 : HCCL_ERROR("tag not found. the fail tag is not from HCCL. tag[%s]", tag.c_str()), false);
771 6 : auto& opDataQueIt = opDataIt->second;
772 6 : CHK_PRT_RET(opDataQueIt->size() == 0, HCCL_ERROR("[TaskExceptionHandler][Callback] OpData queue size 0"), false);
773 2 : auto opDataInfo = opDataQueIt->front();
774 4 : while (opDataQueIt->size() > 0) {
775 2 : HCCL_DEBUG(
776 : "[TaskExceptionHandler][Callback]index %u opData index %u size %u", index, opDataQueIt->front().index,
777 : opDataQueIt->size());
778 2 : if (index == opDataQueIt->front().index) {
779 2 : opDataInfo = opDataQueIt->front();
780 2 : opDataFound = true; // 需要匹配最后下发的task，不能break
781 : }
782 2 : opDataQueIt->pop();
783 : }
784 2 : if (!opDataFound) {
785 0 : return false;
786 : }
787 :
788 2 : PrintOpDataInfo(opDataInfo, isFftsPlus, stageErrInfo);
789 2 : return true;
790 12 : }
791 :
792 12 : bool TaskExceptionHandler::DealExceptionGroupRank(
793 : rtExceptionInfo* exceptionInfo, std::string& tag, bool isFftsPlus, std::string& groupRankContentInfo,
794 : std::string& stageErrInfo)
795 : {
796 12 : std::unique_lock<std::mutex> lock(groupRankMapMutex[exceptionInfo->deviceid]);
797 12 : auto groupRankIt = groupRankMap[exceptionInfo->deviceid].find(tag);
798 12 : CHK_PRT_RET(
799 : groupRankIt == groupRankMap[exceptionInfo->deviceid].end(),
800 : HCCL_INFO("tag not found. the fail tag is not from HCCL. tag[%s]", tag.c_str()), false);
801 :
802 6 : auto groupUdiIt = groupUdiMap[exceptionInfo->deviceid].find(groupRankIt->second.first);
803 6 : CHK_PRT_RET(
804 : groupUdiIt == groupUdiMap[exceptionInfo->deviceid].end(),
805 : HCCL_INFO("group not found. the fail group is not from HCCL. group[%s]", groupRankIt->second.first.c_str()),
806 : false);
807 :
808 6 : string peerRankStr;
809 6 : if ((groupRankIt->second.second)->remoteRankId != INVALID_VALUE_RANKSIZE) {
810 0 : peerRankStr += "], peerRankId[";
811 0 : peerRankStr += std::to_string((groupRankIt->second.second)->remoteRankId);
812 : }
813 :
814 6 : string groupRankContent;
815 6 : groupRankContent += "group:[";
816 6 : groupRankContent += groupRankIt->second.first;
817 6 : groupRankContent += "], user define information[";
818 6 : groupRankContent += groupUdiIt->second;
819 6 : groupRankContent += "], rankSize[";
820 6 : groupRankContent += std::to_string((groupRankIt->second.second)->rankSize);
821 6 : groupRankContent += "], rankId[";
822 6 : groupRankContent += std::to_string((groupRankIt->second.second)->rankId);
823 6 : groupRankContent += peerRankStr;
824 6 : groupRankContent += "]";
825 6 : groupRankContentInfo = groupRankContent;
826 :
827 6 : PrintGroupErrorLog(stageErrInfo, groupRankContent, tag);
828 6 : return true;
829 12 : }
830 :
831 5 : bool TaskExceptionHandler::DealExceptionCtx(rtExceptionInfo* exceptionInfo)
832 : {
833 5 : std::unique_lock<std::mutex> lock(opCtxInfoMutex[exceptionInfo->deviceid]);
834 5 : if (!FindAndValidateContext(exceptionInfo)) {
835 2 : return false;
836 : }
837 :
838 3 : FFTSOpInfo fftsOpInfo;
839 3 : CtxInfo exceptionCtxInfo;
840 3 : std::string stageErrInfo = "";
841 :
842 3 : if (!ProcessContext(exceptionInfo, stageErrInfo, fftsOpInfo, exceptionCtxInfo)) {
843 0 : return false;
844 : }
845 :
846 3 : u32 index = fftsOpInfo.index;
847 6 : std::string groupRankContentInfo = "";
848 3 : std::string tag(fftsOpInfo.tag.get());
849 :
850 3 : DealExceptionGroupRank(exceptionInfo, tag, true, groupRankContentInfo, stageErrInfo);
851 3 : DealExceptionOpData(exceptionInfo, tag, true, index, stageErrInfo);
852 3 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, tag);
853 3 : if (!errMsgFlag_.exchange(true)) {
854 2 : if (exceptionCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT) {
855 17 : RPT_INPUT_ERR(
856 : true, "EI0002",
857 : std::vector<std::string>(
858 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
859 : std::vector<std::string>(
860 : {std::to_string(exceptionCtxInfo.GetCtxRemoteUserRank()),
861 : exceptionCtxInfo.GetCtxBaseInfoStr().c_str(), (exceptionCtxInfo.GetCtxParaInfoStr()).c_str(),
862 : groupRankContentInfo.c_str()}));
863 1 : } else if (
864 1 : exceptionCtxInfo.taskType == TaskType::TASK_SDMA
865 1 : || exceptionCtxInfo.taskType == TaskType::TASK_REDUCE_INLINE) {
866 0 : RPT_INPUT_ERR(
867 : true, "EI0012",
868 : std::vector<std::string>(
869 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
870 : std::vector<std::string>(
871 : {std::to_string(exceptionCtxInfo.GetCtxRemoteUserRank()),
872 : exceptionCtxInfo.GetCtxBaseInfoStr().c_str(), (exceptionCtxInfo.GetCtxParaInfoStr()).c_str(),
873 : groupRankContentInfo.c_str()}));
874 : }
875 : }
876 3 : return true;
877 7 : }
878 :
879 5 : bool TaskExceptionHandler::FindAndValidateContext(rtExceptionInfo* exceptionInfo)
880 : {
881 5 : auto mapIt = opCtxInfo[exceptionInfo->deviceid].find(exceptionInfo->streamid);
882 5 : if (mapIt == opCtxInfo[exceptionInfo->deviceid].end()) {
883 0 : HCCL_RUN_INFO("stream not found. the fail ctx is not from HCCL. streamid[%u]", exceptionInfo->streamid);
884 0 : return false;
885 : }
886 :
887 5 : auto& queIt = mapIt->second;
888 5 : if (queIt->size() == 0) {
889 2 : HCCL_ERROR("[TaskExceptionHandler][Callback] CtxOpInfo queue size 0");
890 2 : return false;
891 : }
892 :
893 3 : if ((*(queIt->front().second)).size() == 0) {
894 0 : HCCL_ERROR("[TaskExceptionHandler][Callback] CtxInfoVector size 0");
895 0 : return false;
896 : }
897 :
898 3 : return true;
899 : }
900 :
901 3 : void TaskExceptionHandler::PrintFftsCtxInfo(FFTSOpInfo& fftsOpInfo)
902 : {
903 : // 按照每个task占用128字节打印ffts的子图信息
904 3 : if (fftsOpInfo.descBuf != nullptr && fftsOpInfo.descBufLen > 0) {
905 1 : HCCL_ERROR(
906 : "==========FftsPlusTask-begin-context, ctx_addr=%p, descBuflen=%u, ctx_num=%lu==========",
907 : fftsOpInfo.descBuf.get(), fftsOpInfo.descBufLen, fftsOpInfo.descBufLen / 128UL);
908 2 : for (uint32_t i = 0U; i < (fftsOpInfo.descBufLen / 128UL); i++) {
909 1 : HCCL_ERROR(
910 : "stream_id=%u, task_id=%u, FftsPlusTask context_id=%u:", fftsOpInfo.streamID, fftsOpInfo.taskID, i);
911 1 : uint32_t* buf = reinterpret_cast<uint32_t*>(fftsOpInfo.descBuf.get()) + (i * 32U);
912 5 : for (uint32_t j = 0U; j < 32U; j += 8) {
913 4 : HCCL_ERROR(
914 : "context_id=%u, buf[%02u-%02u]=%08x %08x %08x %08x %08x %08x %08x %08x.", i, j, (j + 7U), buf[j],
915 : buf[j + 1U], buf[j + 2U], buf[j + 3U], buf[j + 4U], buf[j + 5U], buf[j + 6U], buf[j + 7U]);
916 : }
917 : }
918 1 : HCCL_ERROR("==========FftsPlusTask-end-context==========");
919 : }
920 3 : return;
921 : }
922 :
923 3 : bool TaskExceptionHandler::ProcessContext(
924 : rtExceptionInfo* exceptionInfo, std::string& stageErrInfo, FFTSOpInfo& fftsOpInfo, CtxInfo& exceptionCtxInfo)
925 : {
926 3 : auto mapIt = opCtxInfo[exceptionInfo->deviceid].find(exceptionInfo->streamid);
927 3 : auto& queIt = mapIt->second;
928 3 : fftsOpInfo = *(queIt->front().first);
929 3 : exceptionCtxInfo = (*(queIt->front().second))[0];
930 3 : uint16_t invalidCtxid = 65535;
931 3 : bool ctxFound = false;
932 :
933 3 : while (queIt->size() > 0) {
934 3 : if (exceptionInfo->taskid == queIt->back().first->taskID) {
935 3 : fftsOpInfo = *(queIt->back().first);
936 3 : if (exceptionInfo->expandInfo.u.fftsPlusInfo.contextId == invalidCtxid) {
937 : // 子图任务粒度下，RTS返回的异常task不包含contexId时的处理，约定contextId为65535。只记录算子信息
938 0 : HCCL_WARNING(
939 : "%sTask run failed, invalid contexid,"
940 : "base opInformation is %s",
941 : stageErrInfo.c_str(), fftsOpInfo.GetBaseInfoStr().c_str());
942 3 : } else if (exceptionInfo->expandInfo.u.fftsPlusInfo.contextId >= queIt->back().second->size()) {
943 0 : HCCL_ERROR(
944 : "%sTask run failed, contextId[%u] is out of vector "
945 : "size[%zu], base opInformation is %s",
946 : stageErrInfo.c_str(), exceptionInfo->expandInfo.u.fftsPlusInfo.contextId,
947 : queIt->back().second->size(), fftsOpInfo.GetBaseInfoStr().c_str());
948 : } else {
949 3 : exceptionCtxInfo = (*(queIt->back().second))[exceptionInfo->expandInfo.u.fftsPlusInfo.contextId];
950 3 : ctxFound = true;
951 : }
952 3 : break;
953 : } else {
954 0 : queIt->pop_back();
955 : }
956 : }
957 :
958 : auto logKeywordL2
959 3 : = exceptionCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
960 3 : stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + LOG_KEYWORDS_HOST + "]";
961 :
962 3 : PrintFftsCtxInfo(fftsOpInfo);
963 :
964 3 : if (!ctxFound) {
965 0 : return false;
966 : }
967 :
968 3 : if (exceptionCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT) { // 只在出错task为NotifyWait时打印前序task序列
969 2 : PrintTaskContextInfo(queIt->back().second, exceptionInfo->expandInfo.u.fftsPlusInfo.contextId, stageErrInfo);
970 : }
971 :
972 3 : queIt->clear();
973 :
974 3 : PrintBaseErrorLog(stageErrInfo, fftsOpInfo.GetBaseInfoStr());
975 3 : PrintContextErrorLog(stageErrInfo, exceptionCtxInfo.GetCtxBaseInfoStr());
976 3 : PrintParaErrorLog(stageErrInfo, exceptionCtxInfo.GetCtxParaInfoStr(), std::string(fftsOpInfo.tag.get()));
977 :
978 3 : return true;
979 3 : }
980 :
981 4 : bool TaskExceptionHandler::DealExceptionOp(rtExceptionInfo* exceptionInfo)
982 : {
983 4 : std::unique_lock<std::mutex> lock(opMapMutex[exceptionInfo->deviceid]);
984 4 : bool taskFound = false;
985 4 : auto mapIt = opMap[exceptionInfo->deviceid].find(exceptionInfo->streamid);
986 4 : CHK_PRT_RET(
987 : mapIt == opMap[exceptionInfo->deviceid].end(),
988 : HCCL_RUN_INFO("stream not found. the fail op is not from HCCL. streamid[%u]", exceptionInfo->streamid), false);
989 3 : auto& queIt = mapIt->second;
990 3 : CHK_PRT_RET(queIt->size() == 0, HCCL_ERROR("[TaskExceptionHandler][Callback] OpInfo queue size 0"), false);
991 3 : auto exceptionOpInfo = queIt->back();
992 3 : while (queIt->size() > 0) {
993 3 : if (exceptionInfo->taskid == queIt->back().taskID) {
994 3 : exceptionOpInfo = queIt->back();
995 3 : taskFound = true; // 从后往前匹配最后下发的相同taskId
996 3 : break;
997 : }
998 0 : queIt->pop_back();
999 : }
1000 3 : if (!taskFound) {
1001 0 : return false;
1002 : }
1003 3 : queIt->clear();
1004 :
1005 : auto logKeywordL2
1006 3 : = exceptionInfo->retcode == ACL_ERROR_RT_FFTS_PLUS_TIMEOUT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
1007 3 : auto stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + LOG_KEYWORDS_HOST + "]";
1008 :
1009 3 : PrintBaseErrorLog(stageErrInfo, exceptionOpInfo.GetBaseInfoStr());
1010 3 : u32 index = exceptionOpInfo.index;
1011 6 : std::string groupRankContentInfo = "";
1012 3 : std::string tag(exceptionOpInfo.tag.get());
1013 3 : DealExceptionGroupRank(exceptionInfo, tag, true, groupRankContentInfo, stageErrInfo);
1014 3 : DealExceptionOpData(exceptionInfo, tag, true, index, stageErrInfo);
1015 3 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, tag);
1016 3 : if (!errMsgFlag_.exchange(true)) {
1017 2 : if (exceptionInfo->retcode == ACL_ERROR_RT_FFTS_PLUS_TIMEOUT) {
1018 16 : RPT_INPUT_ERR(
1019 : true, "EI0002",
1020 : std::vector<std::string>(
1021 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1022 : std::vector<std::string>(
1023 : {"unknown", exceptionOpInfo.GetBaseInfoStr().c_str(), errMsg.c_str(),
1024 : groupRankContentInfo.c_str()}));
1025 : }
1026 : }
1027 3 : return true;
1028 6 : }
1029 :
1030 3 : void TaskExceptionHandler::PrintTaskContextInfo(
1031 : const std::shared_ptr<std::deque<TaskInfo>>& taskQue, std::string& stageErrInfo)
1032 : {
1033 3 : HCCL_ERROR(
1034 : "%sTask run failed, context sequence before error task is "
1035 : "[NotifyRecord:NR(rank,id), NotifyWait:NW(rank,id), Memcpy:M(rank), Reduce: R(rank), "
1036 : "InlineReduce:IR(rank), RDMASend:RS(rank,id)]:",
1037 : stageErrInfo.c_str());
1038 3 : std::string taskContextInfo = "";
1039 3 : u32 startIndex = (taskQue->size() > TASK_CONTEXT_SIZE) ? (taskQue->size() - TASK_CONTEXT_SIZE) : 0;
1040 3 : for (; startIndex < taskQue->size(); startIndex++) {
1041 0 : auto taskInfo = taskQue->at(startIndex);
1042 :
1043 0 : std::string taskStr = GetTaskBriefsName(taskInfo.taskType);
1044 0 : taskStr += "(";
1045 0 : taskStr += taskInfo.GetRankInfo();
1046 0 : if (taskInfo.taskType == TaskType::TASK_NOTIFY_RECORD || taskInfo.taskType == TaskType::TASK_NOTIFY_WAIT
1047 0 : || taskInfo.taskType == TaskType::TASK_RDMA) {
1048 0 : taskStr += ("," + taskInfo.GetNotifyInfo());
1049 : }
1050 0 : taskStr += "),";
1051 0 : if (taskContextInfo.size() + taskStr.size() >= TASK_CONTEXT_INFO_SIZE) {
1052 0 : HCCL_ERROR("%s%s ...", stageErrInfo.c_str(), taskContextInfo.c_str());
1053 0 : taskContextInfo = "";
1054 : }
1055 0 : taskContextInfo += taskStr;
1056 0 : }
1057 3 : HCCL_ERROR("%s%s end.", stageErrInfo.c_str(), taskContextInfo.c_str());
1058 6 : return;
1059 3 : }
1060 :
1061 0 : void TaskExceptionHandler::ParseTaskSyncFlag(s32* flagMem, u32 flagMemSize, u32 rankSize, u32 rank, u32 index)
1062 : {
1063 0 : u32 chips1v1 = std::min(rankSize * NUM_BLOCKS_PER_RANK, MAX_RANK_SIZE_SUPERPOD) * NOTIFY_NUM * INTERVAL_1V1;
1064 0 : u32 cores1v1 = MAX_NUM_BLOCKS * NOTIFY_GROUPS_1V1 * INTERVAL_1V1;
1065 0 : u32 chips1vN = PRINT_1VN_NUM * INTERVAL_1VN * NOTIFY_GROUPS_1V1;
1066 0 : u32 cores1vN = PRINT_1VN_NUM * INTERVAL_1VN * NOTIFY_GROUPS_1V1;
1067 0 : u32 chipsNv1 = PRINT_NV1_NUM * INTERVAL_NV1 * NOTIFY_GROUPS_1V1;
1068 0 : u32 coresNv1 = PRINT_NV1_NUM * INTERVAL_NV1 * NOTIFY_GROUPS_1V1;
1069 0 : u32 count = rankSize * CORE_PER_CARDS * INTERVAL_COUNT;
1070 0 : u32 syncCount = (chips1v1 + cores1v1 + chips1vN + cores1vN + chipsNv1 + coresNv1) * PING_PONG_NUM + count;
1071 0 : u32 total = syncCount * sizeof(u32);
1072 0 : if (total > flagMemSize) {
1073 0 : HCCL_ERROR("rank %u opIndex=%u flag mem size %u is too little total %u.", rank, index, flagMemSize, total);
1074 0 : return;
1075 : }
1076 :
1077 0 : s32* buf = flagMem;
1078 0 : u32 offset = 0;
1079 :
1080 0 : const std::string PREFIX[PING_PONG_NUM] = {"ping", "pong"};
1081 0 : std::string str;
1082 0 : for (u32 i = 0; i < PING_PONG_NUM; ++i) {
1083 : // print chips1v1
1084 0 : str = SerializeSyncFlag(buf + offset, rankSize * NUM_BLOCKS_PER_RANK * NOTIFY_NUM, INTERVAL_1V1);
1085 0 : offset += chips1v1;
1086 0 : HCCL_ERROR("rank %u opIndex %u chips 1v1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1087 :
1088 0 : str = SerializeSyncFlag(buf + offset, MAX_NUM_BLOCKS * NOTIFY_GROUPS_1V1, INTERVAL_1V1);
1089 0 : offset += cores1v1;
1090 0 : HCCL_ERROR("rank %u opIndex %u cores 1v1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1091 :
1092 0 : str = SerializeSyncFlag(buf + offset, PRINT_1VN_NUM * NOTIFY_GROUPS_1V1, INTERVAL_1VN);
1093 0 : offset += chips1vN;
1094 0 : HCCL_ERROR("rank %u opIndex %u chips 1vn sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1095 :
1096 0 : str = SerializeSyncFlag(buf + offset, PRINT_1VN_NUM * NOTIFY_GROUPS_1V1, INTERVAL_1VN);
1097 0 : offset += cores1vN;
1098 0 : HCCL_ERROR("rank %u opIndex %u cores 1vn sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1099 :
1100 0 : str = SerializeSyncFlag(buf + offset, PRINT_NV1_NUM * NOTIFY_GROUPS_1V1, INTERVAL_NV1);
1101 0 : offset += chipsNv1;
1102 0 : HCCL_ERROR("rank %u opIndex %u chips nv1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1103 :
1104 0 : str = SerializeSyncFlag(buf + offset, PRINT_NV1_NUM * NOTIFY_GROUPS_1V1, INTERVAL_NV1);
1105 0 : offset += coresNv1;
1106 0 : HCCL_ERROR("rank %u opIndex %u cores nv1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1107 : }
1108 0 : str = SerializeSyncFlag(buf + offset, rankSize * CORE_PER_CARDS, INTERVAL_COUNT);
1109 0 : HCCL_ERROR("rank %u opIndex %u sync count [%s]", rank, index, str.c_str());
1110 0 : }
1111 :
1112 0 : std::string TaskExceptionHandler::SerializeSyncFlag(s32* buf, u32 num, u32 interval)
1113 : {
1114 0 : std::stringstream ss;
1115 0 : s32* pos = buf;
1116 0 : for (u32 i = 0; i < num; i = i + 1) {
1117 0 : ss << std::dec << " " << *pos;
1118 0 : pos = pos + interval;
1119 : }
1120 0 : return ss.str();
1121 0 : }
1122 :
1123 0 : void TaskExceptionHandler::PrintTaskAivBuffer(const std::shared_ptr<std::deque<TaskInfo>>& taskQue)
1124 : {
1125 0 : if (taskQue->empty()) {
1126 0 : return;
1127 : }
1128 : // width参考aiv_communication_base.cc的MAX_FLAG_SIZE_PER_KERNEL
1129 :
1130 0 : u32 flagMemSize = 1024 * 1024;
1131 0 : auto& taskInfo = taskQue->back();
1132 0 : u32 realRankSize = taskInfo.taskPara.Aiv.rankSize;
1133 0 : void* tmpFlagMem = malloc(flagMemSize);
1134 0 : if (tmpFlagMem == nullptr) {
1135 0 : return;
1136 : }
1137 0 : s32* flagMem = static_cast<s32*>(tmpFlagMem);
1138 0 : hrtMemSyncCopy(
1139 0 : flagMem, flagMemSize, reinterpret_cast<u8*>(taskInfo.taskPara.Aiv.flagMem), flagMemSize,
1140 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST);
1141 :
1142 0 : ParseTaskSyncFlag(flagMem, flagMemSize, realRankSize, taskInfo.taskPara.Aiv.rank, taskInfo.index);
1143 0 : free(flagMem);
1144 : }
1145 :
1146 0 : void TaskExceptionHandler::PrintTaskAivInfo(const std::shared_ptr<std::deque<TaskInfo>>& taskQue)
1147 : {
1148 0 : HCCL_ERROR("[PrintTaskAivInfo] print start: ");
1149 : // 从后往前遍历，最多打印PRINT_TASK_AIV_INFO_COUNT个taskAiv
1150 0 : int cnt = PRINT_TASK_AIV_INFO_COUNT;
1151 0 : for (auto it = taskQue->end() - 1; it >= taskQue->begin(); --it) {
1152 0 : if (!it->isAlgInfo) {
1153 0 : continue;
1154 : }
1155 0 : if (cnt <= 0) {
1156 0 : break;
1157 : }
1158 0 : auto taskInfo = *it;
1159 0 : HCCL_ERROR("[AIV](%s) ", taskInfo.GetParaAiv().c_str());
1160 0 : cnt--;
1161 0 : }
1162 0 : HCCL_ERROR("[PrintTaskAivInfo] print end.");
1163 0 : return;
1164 : }
1165 :
1166 5 : void splitAndPrintErrStr(const std::string& s)
1167 : {
1168 5 : std::vector<string> parts;
1169 5 : std::istringstream iss(s);
1170 5 : std::string part;
1171 :
1172 : // 将字符串按照空格分隔
1173 114 : while (iss >> part) {
1174 109 : parts.push_back(part);
1175 : }
1176 :
1177 : // 每10组作为一行打印，暂不做通用化处理
1178 5 : constexpr u32 plen = 10;
1179 19 : for (size_t i = 0; i < parts.size(); i += plen) {
1180 14 : std::string line;
1181 123 : for (size_t j = i; j < i + plen && j < parts.size(); ++j) {
1182 109 : if (j != i) {
1183 95 : line += " ";
1184 : }
1185 109 : line += parts[j];
1186 : }
1187 14 : HCCL_ERROR("%s", line.c_str());
1188 14 : }
1189 5 : }
1190 :
1191 11 : HcclResult TaskExceptionHandler::PrintCommAivInfo()
1192 : {
1193 : /* 本函数的目的：在任务失败后，遍历当前device的所有通信域
1194 : 对于通信域内存在AIV算子的情况进行统计和打印
1195 : 提示用户如果有多个通信域存在AIV算子可能导致执行卡住
1196 : */
1197 11 : u32 groupHasAivCount = 0;
1198 11 : u32 groupNoAivCount = 0;
1199 11 : s32 deviceLogicId = -1;
1200 11 : std::stringstream groupHasAivInfo;
1201 11 : std::stringstream groupNoAivInfo;
1202 :
1203 11 : HcclResult ret = hrtGetDevice(&deviceLogicId);
1204 11 : if (ret != HCCL_SUCCESS) {
1205 1 : HCCL_ERROR("[TaskExceptionHandler][PrintCommAivInfo]hrtGetDevice failed, ret[%d]", ret);
1206 1 : return HCCL_E_PARA;
1207 : }
1208 :
1209 : // 轮询aivGroupIndexMap_[deviceLogicId]的group，确认是否此group内有aiv算子
1210 : // 对于存在aiv算子的，记录和打印group信息和aiv信息
1211 10 : if (aivGroupIndexMap_[deviceLogicId].size() == 0) {
1212 6 : HCCL_ERROR("[TaskExceptionHandler][PrintCommAivInfo] aiv group not record");
1213 6 : return HCCL_SUCCESS;
1214 : }
1215 :
1216 113 : for (auto it = aivGroupIndexMap_[deviceLogicId].begin(); it != aivGroupIndexMap_[deviceLogicId].end(); it++) {
1217 109 : if (it->second == 0) {
1218 5 : groupNoAivInfo << "[" << it->first.c_str() << "] ";
1219 5 : groupNoAivCount++;
1220 : } else {
1221 104 : groupHasAivInfo << "[" << it->first.c_str() << "] ";
1222 104 : groupHasAivCount++;
1223 : }
1224 : }
1225 :
1226 : // 如果遍历发现，存在通信域内执行过aiv算子，则提示有可能有卡死风险；大于0则提示，因为MC2也有可能有aiv算子。
1227 4 : if (groupHasAivCount != 0) {
1228 3 : HCCL_ERROR(
1229 : "[TaskExceptionHandler][PrintCommAivInfo] multi groups include aiv alg, may cause execution stuck."
1230 : " has aiv group count[%u]",
1231 : groupHasAivCount);
1232 3 : HCCL_ERROR("groups has aiv list[groupName]:");
1233 3 : splitAndPrintErrStr(groupHasAivInfo.str());
1234 : }
1235 :
1236 : // 通信域不包含aiv算子的，也一并提示
1237 4 : if (groupNoAivCount != 0) {
1238 2 : HCCL_ERROR("[TaskExceptionHandler][PrintCommAivInfo] no aiv alg group count[%u].", groupNoAivCount);
1239 2 : HCCL_ERROR("groups no aiv list[groupName]: ");
1240 2 : splitAndPrintErrStr(groupNoAivInfo.str());
1241 : }
1242 :
1243 4 : return HCCL_SUCCESS;
1244 11 : }
1245 :
1246 7 : bool TaskExceptionHandler::DealExceptionTask(rtExceptionInfo* exceptionInfo)
1247 : {
1248 7 : std::unique_lock<std::mutex> lock(taskMapMutex[exceptionInfo->deviceid]);
1249 7 : bool taskFound = false;
1250 7 : auto mapIt = taskMap[exceptionInfo->deviceid].find(exceptionInfo->streamid);
1251 7 : CHK_PRT_RET(
1252 : mapIt == taskMap[exceptionInfo->deviceid].end(),
1253 : HCCL_RUN_INFO("stream not found. the fail task is not from HCCL. streamid[%u]", exceptionInfo->streamid),
1254 : false);
1255 6 : auto& queIt = mapIt->second;
1256 6 : CHK_PRT_RET(queIt->size() == 0, HCCL_ERROR("[TaskExceptionHandler][Callback] TaskInfo queue size 0"), false);
1257 :
1258 : // 从后往前匹配最后下发的相同taskId
1259 5 : auto exceptionTaskInfo = queIt->back();
1260 6 : while (queIt->size() > 0) {
1261 6 : if (exceptionInfo->taskid == queIt->back().taskID) {
1262 5 : exceptionTaskInfo = queIt->back();
1263 5 : taskFound = true;
1264 5 : break;
1265 : }
1266 1 : queIt->pop_back();
1267 : }
1268 5 : if (!taskFound) {
1269 0 : return false;
1270 : }
1271 :
1272 : // 检测是否存在多通信域有aiv算子情况，提示可能导致执行卡住
1273 5 : CHK_PRT_RET(PrintCommAivInfo(), HCCL_ERROR("[TaskExceptionHandler] PrintCommAivInfo failed."), false);
1274 :
1275 5 : std::string logKeywordL2;
1276 5 : std::string logKeywordL3;
1277 :
1278 5 : if (exceptionTaskInfo.isAlgInfo) {
1279 : // aiv场景若根据retCode是否为ACL_ERROR_RT_VECTOR_CORE_TIMEOUT判断是否为超时报错
1280 0 : logKeywordL2 = exceptionInfo->retcode == ACL_ERROR_RT_VECTOR_CORE_TIMEOUT ? LOG_KEYWORDS_TIMEOUT :
1281 0 : LOG_KEYWORDS_RUN_FAILED;
1282 0 : logKeywordL3 = LOG_KEYWORDS_AIV;
1283 : } else {
1284 : // 非aiv场景根据当前报错的taskType是否为TASK_NOTIFY_WAIT判断是否为超时报错
1285 : logKeywordL2
1286 5 : = exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
1287 5 : logKeywordL3 = LOG_KEYWORDS_HOST_TS;
1288 : }
1289 :
1290 5 : auto stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + logKeywordL3 + "]";
1291 :
1292 5 : if (exceptionTaskInfo.isAlgInfo) {
1293 0 : PrintTaskAivBuffer(queIt);
1294 0 : PrintTaskAivInfo(queIt);
1295 5 : } else if (exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT) {
1296 3 : queIt->pop_back();
1297 : // 只在出错task为NotifyWait时打印前序task序列
1298 3 : PrintTaskContextInfo(queIt, stageErrInfo);
1299 : }
1300 :
1301 5 : queIt->clear();
1302 5 : HCCL_ERROR("%sTask from HCCL run failed.", stageErrInfo.c_str());
1303 : // 防止tag字符串过长， 信息分开打印
1304 5 : PrintBaseErrorLog(stageErrInfo, exceptionTaskInfo.GetBaseInfoStr());
1305 5 : PrintParaErrorLog(stageErrInfo, exceptionTaskInfo.GetParaInfoStr(), exceptionTaskInfo.tag);
1306 5 : u32 index = exceptionTaskInfo.index;
1307 5 : std::string groupRankContentInfo = "";
1308 5 : if (!exceptionTaskInfo.isAlgInfo) {
1309 : // AlgInfo时不打印group rank等信息
1310 5 : DealExceptionGroupRank(exceptionInfo, exceptionTaskInfo.tag, false, groupRankContentInfo, stageErrInfo);
1311 : }
1312 5 : DealExceptionOpData(exceptionInfo, exceptionTaskInfo.tag, false, index, stageErrInfo);
1313 5 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, exceptionTaskInfo.tag);
1314 5 : if (!errMsgFlag_.exchange(true)) {
1315 4 : if (logKeywordL2 == LOG_KEYWORDS_TIMEOUT) {
1316 34 : RPT_INPUT_ERR(
1317 : true, "EI0002",
1318 : std::vector<std::string>(
1319 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1320 : std::vector<std::string>(
1321 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()), exceptionTaskInfo.GetBaseInfoStr().c_str(),
1322 : (exceptionTaskInfo.GetParaInfoStr()).c_str(), groupRankContentInfo.c_str()}));
1323 : } else {
1324 34 : RPT_INPUT_ERR(
1325 : true, "EI0012",
1326 : std::vector<std::string>(
1327 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1328 : std::vector<std::string>(
1329 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()), exceptionTaskInfo.GetBaseInfoStr().c_str(),
1330 : (exceptionTaskInfo.GetParaInfoStr()).c_str(), groupRankContentInfo.c_str()}));
1331 : }
1332 : }
1333 5 : return true;
1334 15 : }
1335 :
1336 11 : void TaskExceptionHandler::PrintAicpuErrorMessage(rtExceptionInfo* exceptionInfo, bool& isExistAicpuError)
1337 : {
1338 11 : ErrorMessageReport errorMessage;
1339 11 : unique_lock<std::mutex> lock(g_commHadCallbackArrayMutex);
1340 11 : if (g_commHadCallbackArray[exceptionInfo->deviceid]) {
1341 : // 防止同一个device上出现通信主流和kernel流均出现task exception时runtime调用两次callback
1342 : // HDC通道信息不是读清，防止aicpu task exception重复上报
1343 2 : HCCL_WARNING("aicpu error message been reported. deviceid[%u]", exceptionInfo->deviceid);
1344 2 : return;
1345 : }
1346 9 : lock.unlock();
1347 9 : if (g_communicatorCallbackMap[exceptionInfo->deviceid].find(exceptionInfo->streamid)
1348 18 : != g_communicatorCallbackMap[exceptionInfo->deviceid].end()) {
1349 : // 找到对应的通信域，并调用回调函数从HDC通道获取AICPU异常信息
1350 7 : errorMessage = (g_communicatorCallbackMap[exceptionInfo->deviceid])[exceptionInfo->streamid]();
1351 7 : if (strlen(errorMessage.tag) > 0) {
1352 6 : isExistAicpuError = true;
1353 6 : string groupRankContent;
1354 6 : u32 streamId = static_cast<u32>(errorMessage.streamId);
1355 6 : std::string tag = std::string(errorMessage.tag);
1356 6 : u32 index = 0;
1357 : TaskParaNotify para(
1358 6 : static_cast<u64>(errorMessage.notifyId), errorMessage.stage, errorMessage.remoteUserRank);
1359 : TaskInfo exceptionTaskInfo(
1360 6 : streamId, errorMessage.taskId, tag, errorMessage.taskType, errorMessage.algType, index, para);
1361 6 : auto logKeywordL2 = exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT ? LOG_KEYWORDS_TIMEOUT :
1362 6 : LOG_KEYWORDS_RUN_FAILED;
1363 6 : auto stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + LOG_KEYWORDS_AICPU + "]";
1364 6 : HCCL_ERROR("%sTask from HCCL run failed.", stageErrInfo.c_str());
1365 : // 防止tag字符串过长， 信息分开打印
1366 6 : PrintBaseErrorLog(stageErrInfo, exceptionTaskInfo.GetBaseInfoStr());
1367 6 : PrintParaErrorLog(stageErrInfo, exceptionTaskInfo.GetParaInfoStr(), exceptionTaskInfo.tag);
1368 6 : PrintGroupErrorMessage(errorMessage, exceptionTaskInfo, groupRankContent, stageErrInfo);
1369 6 : PrintOpDataErrorMessage(exceptionInfo->deviceid, errorMessage, stageErrInfo);
1370 6 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, tag);
1371 6 : if (!errMsgFlag_.exchange(true)) {
1372 3 : if (exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT) {
1373 34 : RPT_INPUT_ERR(
1374 : true, "EI0002",
1375 : std::vector<std::string>(
1376 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1377 : std::vector<std::string>(
1378 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()),
1379 : exceptionTaskInfo.GetBaseInfoStr().c_str(), (exceptionTaskInfo.GetParaInfoStr()).c_str(),
1380 : "none"}));
1381 1 : } else if (
1382 1 : exceptionTaskInfo.taskType == TaskType::TASK_SDMA
1383 0 : || exceptionTaskInfo.taskType == TaskType::TASK_REDUCE_INLINE) {
1384 18 : RPT_INPUT_ERR(
1385 : true, "EI0012",
1386 : std::vector<std::string>(
1387 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1388 : std::vector<std::string>(
1389 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()),
1390 : exceptionTaskInfo.GetBaseInfoStr().c_str(),
1391 : (exceptionTaskInfo.GetParaInfoStr() + errMsg).c_str(), groupRankContent.c_str()}));
1392 : }
1393 : }
1394 6 : lock.lock();
1395 6 : g_commHadCallbackArray[exceptionInfo->deviceid] = true;
1396 6 : }
1397 : } else {
1398 2 : HCCL_INFO("PrintAicpuErrorMessage streamId[%d] is not found.", exceptionInfo->streamid);
1399 : }
1400 9 : return;
1401 18 : }
1402 :
1403 6 : void TaskExceptionHandler::PrintGroupErrorMessage(
1404 : ErrorMessageReport& errorMessage, TaskInfo& exceptionTaskInfo, string& groupRankContent, string& stageErrInfo)
1405 : {
1406 6 : std::string groupUdi;
1407 6 : std::string groupName = std::string(errorMessage.group);
1408 6 : ProfilerBase::GetUdiByGroup(groupName, groupUdi);
1409 :
1410 6 : groupRankContent += "group:[";
1411 6 : groupRankContent += std::string(errorMessage.group);
1412 6 : groupRankContent += "], user define information[";
1413 6 : groupRankContent += groupUdi;
1414 6 : groupRankContent += "], rankSize[";
1415 6 : groupRankContent += std::to_string(errorMessage.rankSize);
1416 6 : groupRankContent += "], rankId[";
1417 6 : groupRankContent += std::to_string(errorMessage.rankId);
1418 6 : groupRankContent += " ";
1419 6 : groupRankContent += std::to_string(errorMessage.remoteUserRank);
1420 6 : groupRankContent += "]";
1421 :
1422 6 : PrintGroupErrorLog(stageErrInfo, groupRankContent, exceptionTaskInfo.tag);
1423 12 : return;
1424 6 : }
1425 :
1426 6 : void TaskExceptionHandler::PrintOpDataErrorMessage(u32 deviceId, ErrorMessageReport& errorMessage, string& stageErrInfo)
1427 : {
1428 6 : stringstream opDataStr;
1429 6 : opDataStr << "src" << "[0x" << std::hex << errorMessage.srcAddr << "], dst[0x" << std::hex << errorMessage.dstAddr
1430 6 : << "], ";
1431 :
1432 6 : string opStr;
1433 6 : if (errorMessage.reduceType != HcclReduceOp::HCCL_REDUCE_RESERVED) {
1434 0 : opStr += "reduceType[";
1435 0 : opStr += GetReduceOpEnumStr(static_cast<HcclReduceOp>(errorMessage.reduceType));
1436 0 : opStr += "], ";
1437 : }
1438 :
1439 6 : string opDataContent;
1440 6 : opDataContent += "deviceId:[";
1441 6 : opDataContent += std::to_string(deviceId);
1442 6 : opDataContent += "], index[";
1443 6 : opDataContent += std::to_string(errorMessage.opIndex);
1444 6 : opDataContent += "], count[";
1445 6 : opDataContent += std::to_string(errorMessage.count);
1446 6 : opDataContent += "], ";
1447 6 : opDataContent += opStr;
1448 6 : opDataContent += opDataStr.str();
1449 6 : opDataContent += "dataType[";
1450 6 : opDataContent += GetDataTypeEnumStr(errorMessage.dataType);
1451 6 : opDataContent += "].";
1452 :
1453 6 : PrintOpDataErrorLog(stageErrInfo, opDataContent);
1454 12 : return;
1455 6 : }
1456 :
1457 2 : void TaskExceptionHandler::Callback(rtExceptionInfo* exceptionInfo)
1458 : {
1459 2 : HCCL_RUN_INFO("[TaskExceptionHandler][%s]begin to execute hccl task exception callback function.", __func__);
1460 2 : bool isExistAicpuError = false;
1461 2 : if (exceptionInfo == nullptr) {
1462 0 : HCCL_ERROR("[TaskExceptionHandler][Callback] exceptionInfo is nullptr.");
1463 0 : return;
1464 : }
1465 :
1466 2 : PrintAicpuErrorMessage(exceptionInfo, isExistAicpuError);
1467 2 : if (isExistAicpuError) {
1468 : // 如果已经有AICPU上报的task exception, 则host侧无需再次重复上报
1469 0 : return;
1470 : }
1471 : u32 maxDeviceNum;
1472 2 : HcclResult ret = GetMaxDevNum(maxDeviceNum);
1473 2 : if (ret != HCCL_SUCCESS) {
1474 0 : HCCL_ERROR("[GetMaxDevNum] get maxDeviceNum error");
1475 0 : return;
1476 : }
1477 2 : CHK_PRT_RET(
1478 : exceptionInfo->deviceid >= maxDeviceNum,
1479 : HCCL_WARNING(
1480 : "deviceID[%u] from exceptionInfo is bigger than maxDeviceNum[%u]", exceptionInfo->deviceid,
1481 : maxDeviceNum), );
1482 2 : SaluSleep(ONE_MILLISECOND_OF_USLEEP); // sleep 1ms，等待task被存入数据结构
1483 2 : HCCL_DEBUG(
1484 : "[TaskExceptionHandler][Callback]Task run failed, ffts+ task type:%d, TaskExceptionSwitch:%u",
1485 : exceptionInfo->expandInfo.type, GetExternalInputTaskExceptionSwitch());
1486 2 : if (exceptionInfo->expandInfo.type == RT_EXCEPTION_FFTS_PLUS) {
1487 1 : if (GetExternalInputTaskExceptionSwitch() == 1) {
1488 1 : DealExceptionCtx(exceptionInfo); // 子任务粒度
1489 : } else {
1490 0 : DealExceptionOp(exceptionInfo); // 算子粒度
1491 : }
1492 : } else {
1493 1 : DealExceptionTask(exceptionInfo);
1494 : }
1495 2 : return;
1496 : }
1497 506 : HcclResult TaskExceptionHandler::Init()
1498 : {
1499 506 : if (communicatorCount_.fetch_add(1) == 0) {
1500 182 : HCCL_RUN_INFO("[TaskExceptionHandler][%s] register taskFailCallback", __func__);
1501 182 : CHK_RET(hrtRegTaskFailCallbackByModule(Callback));
1502 182 : CHK_RET(hrtGetStreamAvailableNum(maxStrCount));
1503 182 : maxStrCount = (maxStrCount < STREAM_COUNT_UPPER_LIMIT) ? maxStrCount : STREAM_COUNT_UPPER_LIMIT;
1504 : }
1505 506 : maxTaskCount = TASK_COUNT_UPPER_LIMIT;
1506 : // 单算子模式task过多的特殊处理
1507 506 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
1508 449 : maxTaskCount = TASK_COUNT_UPPER_LIMIT_OP_BASE;
1509 : }
1510 :
1511 506 : HCCL_INFO("get from RTS the max stream count[%u] the max task count[%u]", maxStrCount, maxTaskCount);
1512 :
1513 621 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1514 621 : && GetExternalInputTaskExceptionSwitch() == 1) {
1515 330 : for (std::vector<CtxInfo>& ctxInfoVector : ctxInfoArray) {
1516 325 : ctxInfoVector.reserve(100); // vector预留100个ctxInfo空间
1517 : }
1518 : }
1519 :
1520 : // 对全局变量g_commHadCallbackArray进行初始化
1521 33396 : for (u32 i = 0; i < MAX_MODULE_DEVICE_NUM; i++) {
1522 32890 : g_commHadCallbackArray[i] = false;
1523 : }
1524 506 : return HCCL_SUCCESS;
1525 : }
1526 :
1527 808 : HcclResult TaskExceptionHandler::DeInit()
1528 : {
1529 808 : if (communicatorCount_.fetch_sub(1) == 1) {
1530 181 : CHK_RET(hrtRegTaskFailCallbackByModule(nullptr));
1531 181 : HCCL_RUN_INFO("deInit taskFailCallback");
1532 : }
1533 808 : return HCCL_SUCCESS;
1534 : }
1535 :
1536 7 : bool IsOneSideTask(u32 streamId)
1537 : {
1538 7 : std::string tag;
1539 7 : CHK_PRT(ProfilerBase::GetTagByStream(streamId, tag));
1540 7 : if (tag.find("BatchPut_") != std::string::npos || tag.find("BatchGet_") != std::string::npos) {
1541 0 : return true;
1542 : }
1543 7 : return false;
1544 7 : }
1545 :
1546 2 : HcclResult TaskExceptionHandler::Save(
1547 : u32 captureStreamID, u32 streamID, u32 taskID, TaskType& taskType, const TaskParaNotify& para)
1548 : {
1549 : u32 maxDeviceNum;
1550 2 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1551 2 : CHK_PRT_RET(
1552 : deviceLogicId_ >= maxDeviceNum,
1553 : HCCL_ERROR(
1554 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1555 : maxDeviceNum),
1556 : HCCL_E_INTERNAL);
1557 2 : HCCL_INFO(
1558 : "[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u], taskType[%d]", __func__, streamID, taskID,
1559 : taskType);
1560 4 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1561 4 : && GetExternalInputTaskExceptionSwitch() == 1 && !IsOneSideTask(captureStreamID)) {
1562 2 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1563 2 : CtxInfo tmpCtxInfo(taskType, para);
1564 2 : ctxInfoArray[deviceLogicId_].insert(ctxInfoArray[deviceLogicId_].end(), tmpCtxInfo);
1565 2 : return HCCL_SUCCESS;
1566 2 : }
1567 :
1568 0 : std::string tag;
1569 0 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1570 0 : AlgType algType = AlgType::Reserved();
1571 0 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1572 0 : u32 index = 0;
1573 0 : ProfilerBase::GetSubmittedOpCnt(index);
1574 :
1575 0 : TaskInfo tmpTaskInfo(streamID, taskID, tag, taskType, algType, index, para);
1576 0 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1577 :
1578 0 : CHK_RET(InsertRankInfo(tag));
1579 0 : CHK_RET(InsertOpData(tag));
1580 0 : return HCCL_SUCCESS;
1581 0 : }
1582 :
1583 2 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, TaskType& taskType, const TaskParaNotify& para)
1584 : {
1585 2 : return Save(streamID, streamID, taskID, taskType, para);
1586 : }
1587 :
1588 : HcclResult
1589 37 : TaskExceptionHandler::Save(u32 captureStreamID, u32 streamID, u32 taskID, TaskType& taskType, const TaskParaDMA& para)
1590 : {
1591 : u32 maxDeviceNum;
1592 37 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1593 37 : CHK_PRT_RET(
1594 : deviceLogicId_ >= maxDeviceNum,
1595 : HCCL_ERROR(
1596 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1597 : maxDeviceNum),
1598 : HCCL_E_INTERNAL);
1599 37 : HCCL_INFO(
1600 : "[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u], taskType[%d]", __func__, streamID, taskID,
1601 : taskType);
1602 46 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1603 46 : && GetExternalInputTaskExceptionSwitch() == 1 && !IsOneSideTask(captureStreamID)) {
1604 5 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1605 5 : CtxInfo tmpCtxInfo(taskType, para);
1606 5 : ctxInfoArray[deviceLogicId_].insert(ctxInfoArray[deviceLogicId_].end(), tmpCtxInfo);
1607 5 : return HCCL_SUCCESS;
1608 5 : }
1609 :
1610 32 : std::string tag;
1611 32 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1612 32 : AlgType algType = AlgType::Reserved();
1613 32 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1614 32 : u32 index = 0;
1615 32 : ProfilerBase::GetSubmittedOpCnt(index);
1616 :
1617 32 : TaskInfo tmpTaskInfo(streamID, taskID, tag, taskType, algType, index, para);
1618 32 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1619 0 : CHK_RET(InsertRankInfo(tag));
1620 0 : CHK_RET(InsertOpData(tag));
1621 0 : return HCCL_SUCCESS;
1622 32 : }
1623 :
1624 37 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, TaskType& taskType, const TaskParaDMA& para)
1625 : {
1626 37 : return Save(streamID, streamID, taskID, taskType, para);
1627 : }
1628 :
1629 9 : HcclResult TaskExceptionHandler::Save(
1630 : u32 captureStreamID, u32 streamID, u32 taskID, TaskType& taskType, const TaskParaReduce& para)
1631 : {
1632 : u32 maxDeviceNum;
1633 9 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1634 9 : CHK_PRT_RET(
1635 : deviceLogicId_ >= maxDeviceNum,
1636 : HCCL_ERROR(
1637 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1638 : maxDeviceNum),
1639 : HCCL_E_INTERNAL);
1640 9 : HCCL_INFO(
1641 : "[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u], taskType[%d]", __func__, streamID, taskID,
1642 : taskType);
1643 9 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1644 9 : && GetExternalInputTaskExceptionSwitch() == 1 && !IsOneSideTask(captureStreamID)) {
1645 0 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1646 0 : CtxInfo tmpCtxInfo(taskType, para);
1647 0 : ctxInfoArray[deviceLogicId_].insert(ctxInfoArray[deviceLogicId_].end(), tmpCtxInfo);
1648 0 : return HCCL_SUCCESS;
1649 0 : }
1650 :
1651 9 : std::string tag;
1652 9 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1653 9 : AlgType algType = AlgType::Reserved();
1654 9 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1655 9 : u32 index = 0;
1656 9 : ProfilerBase::GetSubmittedOpCnt(index);
1657 :
1658 9 : TaskInfo tmpTaskInfo(streamID, taskID, tag, taskType, algType, index, para);
1659 9 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1660 0 : CHK_RET(InsertRankInfo(tag));
1661 0 : CHK_RET(InsertOpData(tag));
1662 0 : return HCCL_SUCCESS;
1663 9 : }
1664 :
1665 5 : HcclResult TaskExceptionHandler::Save(u32 captureStreamID, u32 streamID, u32 taskID, const TaskParaAiv& para)
1666 : {
1667 : u32 maxDeviceNum;
1668 5 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1669 5 : CHK_PRT_RET(
1670 : deviceLogicId_ >= maxDeviceNum,
1671 : HCCL_ERROR(
1672 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1673 : maxDeviceNum),
1674 : HCCL_E_INTERNAL);
1675 :
1676 5 : std::string tag;
1677 5 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1678 5 : u32 index = 0;
1679 5 : ProfilerBase::GetSubmittedOpCnt(index);
1680 5 : TaskInfo tmpTaskInfo(streamID, taskID, tag, para);
1681 5 : tmpTaskInfo.index = index;
1682 5 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1683 0 : CHK_RET(InsertRankInfo(tag));
1684 0 : CHK_RET(InsertOpData(tag));
1685 0 : return HCCL_SUCCESS;
1686 5 : }
1687 :
1688 2 : HcclResult TaskExceptionHandler::Save(u32 streamID, u32 taskID, const TaskParaAiv& para)
1689 : {
1690 2 : return Save(streamID, streamID, taskID, para);
1691 : }
1692 :
1693 9 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, TaskType& taskType, const TaskParaReduce& para)
1694 : {
1695 9 : return Save(streamID, streamID, taskID, taskType, para);
1696 : }
1697 :
1698 : HcclResult
1699 6 : TaskExceptionHandler::Save(u32 captureStreamID, u32 streamID, u32 taskID, const void* descBuf, size_t descBufLen)
1700 : {
1701 : u32 maxDeviceNum;
1702 6 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1703 6 : CHK_PRT_RET(
1704 : deviceLogicId_ >= maxDeviceNum,
1705 : HCCL_ERROR(
1706 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1707 : maxDeviceNum),
1708 : HCCL_E_INTERNAL);
1709 6 : HCCL_INFO("[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u]", __func__, streamID, taskID);
1710 6 : std::string tag;
1711 6 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1712 6 : AlgType algType = AlgType::Reserved();
1713 6 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1714 6 : u32 index = 0;
1715 6 : ProfilerBase::GetSubmittedOpCnt(index);
1716 :
1717 6 : if (GetExternalInputTaskExceptionSwitch() == 1) {
1718 3 : CHK_RET(InsertOpCtxInfo(streamID, taskID, tag, algType, index, descBuf, descBufLen));
1719 : } else {
1720 3 : CHK_RET(InsertOpMap(streamID, taskID, tag, algType, index));
1721 : }
1722 6 : CHK_RET(InsertRankInfo(tag));
1723 6 : CHK_RET(InsertOpData(tag));
1724 6 : return HCCL_SUCCESS;
1725 6 : }
1726 :
1727 6 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, const void* descBuf, size_t descBufLen)
1728 : {
1729 6 : return Save(streamID, streamID, taskID, descBuf, descBufLen);
1730 : }
1731 :
1732 0 : HcclResult TaskExceptionHandler::SaveToLog(const TaskParaHost& paraHost)
1733 : {
1734 : (void)paraHost;
1735 0 : return HCCL_SUCCESS;
1736 : }
1737 :
1738 52 : HcclResult TaskExceptionHandler::InsertTaskMap(u32& streamID, TaskInfo& tmpTaskInfo) const
1739 : {
1740 52 : std::unique_lock<std::mutex> lock(taskMapMutex[deviceLogicId_]);
1741 52 : auto it = taskMap[deviceLogicId_].find(streamID);
1742 52 : if (it == taskMap[deviceLogicId_].end()) {
1743 : // streamID 复用且不会超过最大stream数量，因此Map的size超过最大stream数量属于异常场景
1744 47 : HCCL_INFO(
1745 : "streamID is [%u], deviceLogicId is [%u], taskMap size is [%u]", streamID, deviceLogicId_,
1746 : taskMap[deviceLogicId_].size());
1747 47 : CHK_PRT_RET(
1748 : taskMap[deviceLogicId_].size() >= maxStrCount,
1749 : HCCL_ERROR(
1750 : "[Insert][TaskMap]taskMap size is "
1751 : "bigger than max stream count[%u]. stream add fail",
1752 : maxStrCount),
1753 : HCCL_E_INTERNAL);
1754 1 : std::shared_ptr<deque<TaskInfo>> tmpTaskInfoQue = nullptr;
1755 1 : EXCEPTION_CATCH((tmpTaskInfoQue = make_shared<deque<TaskInfo>>()), return HCCL_E_PTR);
1756 1 : tmpTaskInfoQue->push_back(tmpTaskInfo);
1757 1 : taskMap[deviceLogicId_].insert({streamID, tmpTaskInfoQue});
1758 1 : } else { // 由于不允许多线程对同一stream操作，因此此处不需要保留锁，并且此处访问量最多，性能考虑也最好不要加锁
1759 5 : lock.unlock();
1760 5 : it->second->push_back(tmpTaskInfo);
1761 5 : if (it->second->size() > maxTaskCount) {
1762 0 : it->second->pop_front();
1763 : }
1764 : }
1765 6 : return HCCL_SUCCESS;
1766 52 : }
1767 : HcclResult
1768 3 : TaskExceptionHandler::InsertOpMap(u32& streamID, u32& taskID, string& tag, AlgType& algType, u32& index) const
1769 : {
1770 3 : FFTSOpInfo tmpOpPara;
1771 6 : char* tmpAddr = new (std::nothrow) char[tag.size() + 1]();
1772 3 : CHK_PTR_NULL(tmpAddr);
1773 3 : tmpOpPara.tag.reset(tmpAddr, default_delete<char[]>());
1774 3 : CHK_SAFETY_FUNC_RET(memcpy_sp(tmpOpPara.tag.get(), tag.size() + 1, tag.data(), tag.size()));
1775 3 : tmpOpPara.streamID = streamID;
1776 3 : tmpOpPara.taskID = taskID;
1777 3 : tmpOpPara.algType = algType;
1778 3 : tmpOpPara.index = index;
1779 3 : std::unique_lock<std::mutex> lock(opMapMutex[deviceLogicId_]); // 防止存入和读取冲突
1780 3 : auto it = opMap[deviceLogicId_].find(streamID);
1781 3 : if (it == opMap[deviceLogicId_].end()) {
1782 1 : CHK_PRT_RET(
1783 : opMap[deviceLogicId_].size() >= maxStrCount,
1784 : HCCL_ERROR(
1785 : "[Insert][OpMap]Map size is "
1786 : "bigger than max stream count[%u]. stream add fail",
1787 : maxStrCount),
1788 : HCCL_E_INTERNAL);
1789 1 : std::shared_ptr<deque<FFTSOpInfo>> tmpOpInfoQue = nullptr;
1790 1 : EXCEPTION_CATCH((tmpOpInfoQue = make_shared<deque<FFTSOpInfo>>()), return HCCL_E_PTR);
1791 1 : tmpOpInfoQue->push_back(tmpOpPara);
1792 1 : opMap[deviceLogicId_].insert({streamID, tmpOpInfoQue});
1793 1 : } else {
1794 2 : it->second->push_back(tmpOpPara);
1795 2 : if (it->second->size() > maxTaskCount) {
1796 0 : it->second->pop_front();
1797 : }
1798 : }
1799 3 : return HCCL_SUCCESS;
1800 3 : }
1801 3 : HcclResult TaskExceptionHandler::InsertOpCtxInfo(
1802 : u32& streamID, u32& taskID, string& tag, AlgType& algType, u32& index, const void* descBuf, size_t descBufLen) const
1803 : {
1804 3 : FFTSOpInfo tmpOpInfo;
1805 6 : char* tmpAddr = new (std::nothrow) char[tag.size() + 1]();
1806 3 : CHK_PTR_NULL(tmpAddr);
1807 3 : tmpOpInfo.tag.reset(tmpAddr, default_delete<char[]>());
1808 3 : CHK_SAFETY_FUNC_RET(memcpy_sp(tmpOpInfo.tag.get(), tag.size() + 1, tag.data(), tag.size()));
1809 3 : tmpOpInfo.streamID = streamID;
1810 3 : tmpOpInfo.taskID = taskID;
1811 3 : tmpOpInfo.algType = algType;
1812 3 : tmpOpInfo.index = index;
1813 3 : if (descBuf != nullptr && descBufLen > 0) {
1814 130 : char* tmpDescBuf = new (std::nothrow) char[descBufLen + 1]();
1815 1 : CHK_PTR_NULL(tmpDescBuf);
1816 1 : tmpOpInfo.descBuf.reset(tmpDescBuf, default_delete<char[]>());
1817 1 : CHK_SAFETY_FUNC_RET(memcpy_sp(tmpOpInfo.descBuf.get(), descBufLen + 1, descBuf, descBufLen));
1818 1 : tmpOpInfo.descBufLen = descBufLen;
1819 : }
1820 3 : std::shared_ptr<FFTSOpInfo> tmpOpInfoPtr = nullptr;
1821 3 : EXCEPTION_CATCH((tmpOpInfoPtr = std::make_shared<FFTSOpInfo>()), return HCCL_E_PTR);
1822 3 : *tmpOpInfoPtr = tmpOpInfo;
1823 3 : std::shared_ptr<vector<CtxInfo>> tempCtxVectorPtr = nullptr;
1824 3 : EXCEPTION_CATCH((tempCtxVectorPtr = std::make_shared<vector<CtxInfo>>()), return HCCL_E_PTR);
1825 3 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1826 3 : *tempCtxVectorPtr = ctxInfoArray[deviceLogicId_];
1827 3 : auto tempPair = std::make_pair(tmpOpInfoPtr, tempCtxVectorPtr);
1828 3 : std::unique_lock<std::mutex> infoLock(opCtxInfoMutex[deviceLogicId_]); // 防止存入和读取冲突
1829 3 : auto tempDeque = opCtxInfo[deviceLogicId_].find(streamID);
1830 3 : if (tempDeque == opCtxInfo[deviceLogicId_].end()) {
1831 2 : CHK_PRT_RET(
1832 : opCtxInfo[deviceLogicId_].size() >= maxStrCount,
1833 : HCCL_ERROR(
1834 : "[Insert][opCtxInfo]Map size is "
1835 : "bigger than max stream count[%u]. stream add fail",
1836 : maxStrCount),
1837 : HCCL_E_INTERNAL);
1838 : std::shared_ptr<std::deque<std::pair<std::shared_ptr<FFTSOpInfo>, std::shared_ptr<std::vector<CtxInfo>>>>>
1839 2 : tmpOpInfoQue = nullptr;
1840 2 : EXCEPTION_CATCH(
1841 : (tmpOpInfoQue = std::make_shared<
1842 : std::deque<std::pair<std::shared_ptr<FFTSOpInfo>, std::shared_ptr<std::vector<CtxInfo>>>>>()),
1843 : return HCCL_E_PTR);
1844 2 : tmpOpInfoQue->push_back(tempPair);
1845 2 : opCtxInfo[deviceLogicId_].insert({streamID, tmpOpInfoQue});
1846 2 : } else {
1847 1 : tempDeque->second->push_back(tempPair);
1848 1 : if (tempDeque->second->size() > maxTaskCount) {
1849 0 : tempDeque->second->pop_front();
1850 : }
1851 : }
1852 3 : ctxInfoArray[deviceLogicId_].clear();
1853 3 : return HCCL_SUCCESS;
1854 3 : }
1855 :
1856 6 : HcclResult TaskExceptionHandler::InsertRankInfo(std::string& tag) const
1857 : {
1858 6 : std::string groupName;
1859 6 : CHK_RET(ProfilerBase::GetGroupNameByTag(tag, groupName));
1860 6 : GroupRankInfo groupRankInfo;
1861 6 : CHK_RET(ProfilerBase::GetRankInfoByGroup(groupName, groupRankInfo));
1862 6 : std::string groupUdi;
1863 6 : CHK_RET(ProfilerBase::GetUdiByGroup(groupName, groupUdi));
1864 :
1865 6 : HCCL_DEBUG("[TaskExceptionHandler][Callback]InsertRankInfo tag %s group %s", tag.c_str(), groupName.c_str());
1866 : {
1867 6 : std::unique_lock<std::mutex> groupRankMapLock(groupRankMapMutex[deviceLogicId_]);
1868 6 : std::shared_ptr<GroupRankInfo> tmpRankInfo = nullptr;
1869 6 : EXCEPTION_CATCH((tmpRankInfo = std::make_shared<GroupRankInfo>()), return HCCL_E_PTR);
1870 6 : *tmpRankInfo = groupRankInfo;
1871 6 : auto groupRankIt = groupRankMap[deviceLogicId_].find(tag);
1872 6 : if (groupRankIt == groupRankMap[deviceLogicId_].end()) {
1873 2 : auto tempPair = std::make_pair(groupName, tmpRankInfo);
1874 2 : groupRankMap[deviceLogicId_].insert({tag, tempPair});
1875 2 : } else {
1876 4 : groupRankIt->second.second = tmpRankInfo;
1877 : }
1878 6 : }
1879 :
1880 : {
1881 6 : std::lock_guard<std::mutex> groupUdiMapLock(groupUdiMapMutex[deviceLogicId_]);
1882 6 : auto groupUdiIt = groupUdiMap[deviceLogicId_].find(groupName);
1883 6 : if (groupUdiIt == groupUdiMap[deviceLogicId_].end()) {
1884 2 : groupUdiMap[deviceLogicId_].insert({groupName, groupUdi});
1885 : } else {
1886 4 : groupUdiIt->second = groupUdi;
1887 : }
1888 6 : }
1889 :
1890 6 : return HCCL_SUCCESS;
1891 6 : }
1892 :
1893 6 : HcclResult TaskExceptionHandler::InsertOpData(std::string& tag) const
1894 : {
1895 6 : OpDataInfo opDataInfo;
1896 6 : CHK_RET(ProfilerBase::GetOpDataInfoByTag(tag, opDataInfo));
1897 6 : std::unique_lock<std::mutex> lock(tagOpDataMapMutex[deviceLogicId_]);
1898 6 : auto tempDeque = tagOpDataMap[deviceLogicId_].find(tag);
1899 6 : if (tempDeque == tagOpDataMap[deviceLogicId_].end()) {
1900 2 : std::shared_ptr<queue<OpDataInfo>> tmpOpDataInfo = nullptr;
1901 2 : EXCEPTION_CATCH((tmpOpDataInfo = std::make_shared<queue<OpDataInfo>>()), return HCCL_E_PTR);
1902 2 : tmpOpDataInfo->push(opDataInfo);
1903 2 : tagOpDataMap[deviceLogicId_].insert({tag, tmpOpDataInfo});
1904 2 : HCCL_DEBUG("[TaskExceptionHandler][Callback]InsertOpData index %u tag %s", opDataInfo.index, tag.c_str());
1905 2 : } else {
1906 4 : HCCL_DEBUG(
1907 : "[TaskExceptionHandler][Callback]InsertOpData index %u opData index %u size %u tag %s", opDataInfo.index,
1908 : tempDeque->second->back().index, (tempDeque->second)->size(), tag.c_str());
1909 4 : if (tempDeque->second->back().index != opDataInfo.index) { // 需要去重，taskid不同时可能是同一个
1910 0 : tempDeque->second->push(opDataInfo);
1911 : }
1912 4 : if ((tempDeque->second)->size() > 3000) { // 队列深度大于3000则老化
1913 0 : HCCL_DEBUG("[Insert][opDataMap]Map size is [%u], need to pop head data.", (tempDeque->second)->size());
1914 0 : tempDeque->second->pop();
1915 : }
1916 : }
1917 6 : return HCCL_SUCCESS;
1918 6 : }
1919 :
1920 13 : HcclResult TaskExceptionHandler::Flush() { return HCCL_SUCCESS; }
1921 :
1922 0 : HcclResult TaskExceptionHandler::TaskExceptionHandler::Run(const StepData& stepData)
1923 : {
1924 : (void)stepData;
1925 0 : return HCCL_SUCCESS;
1926 : }
|
