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