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 47 : void RegisterGetErrStatusVecCallBack(GetErrStatusVecCallBack p1)
36 : {
37 47 : g_GetErrStatusVecCallBack = p1;
38 47 : 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 = {};
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(
731 : OpDataInfo& opDataInfo, [[maybe_unused]] bool isFftsPlus, std::string& stageErrInfo)
732 : {
733 2 : stringstream opDataStr;
734 2 : opDataStr << "src" << "[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(opDataInfo.src))
735 2 : << "], dst[0x" << std::hex << static_cast<u64>(reinterpret_cast<uintptr_t>(opDataInfo.dst)) << "], ";
736 :
737 2 : string opStr;
738 2 : if (opDataInfo.reduceType != HcclReduceOp::HCCL_REDUCE_RESERVED) {
739 0 : opStr += "reduceType[";
740 0 : opStr += GetReduceOpEnumStr(opDataInfo.reduceType);
741 0 : opStr += "], ";
742 : }
743 :
744 2 : string opDataContent;
745 2 : TimeStruct2Str(opDataInfo.tv, opDataContent);
746 2 : opDataContent += ", deviceId[";
747 2 : opDataContent += std::to_string(opDataInfo.deviceId);
748 2 : opDataContent += "], index[";
749 2 : opDataContent += std::to_string(opDataInfo.index);
750 2 : opDataContent += "], count[";
751 2 : opDataContent += std::to_string(opDataInfo.count);
752 2 : opDataContent += "], ";
753 2 : opDataContent += opStr;
754 2 : opDataContent += opDataStr.str();
755 2 : opDataContent += "dataType[";
756 2 : opDataContent += GetDataTypeEnumStr(opDataInfo.dataType);
757 2 : opDataContent += "].";
758 :
759 2 : PrintOpDataErrorLog(stageErrInfo, opDataContent);
760 4 : return;
761 2 : }
762 :
763 12 : bool TaskExceptionHandler::DealExceptionOpData(
764 : rtExceptionInfo* exceptionInfo, std::string& tag, bool isFftsPlus, u32 index, std::string& stageErrInfo)
765 : {
766 12 : bool opDataFound = false;
767 12 : std::unique_lock<std::mutex> lock(tagOpDataMapMutex[exceptionInfo->deviceid]);
768 12 : auto opDataIt = tagOpDataMap[exceptionInfo->deviceid].find(tag);
769 12 : CHK_PRT_RET(
770 : opDataIt == tagOpDataMap[exceptionInfo->deviceid].end(),
771 : HCCL_ERROR("tag not found. the fail tag is not from HCCL. tag[%s]", tag.c_str()), false);
772 6 : auto& opDataQueIt = opDataIt->second;
773 6 : CHK_PRT_RET(opDataQueIt->size() == 0, HCCL_ERROR("[TaskExceptionHandler][Callback] OpData queue size 0"), false);
774 2 : auto opDataInfo = opDataQueIt->front();
775 4 : while (opDataQueIt->size() > 0) {
776 2 : HCCL_DEBUG(
777 : "[TaskExceptionHandler][Callback]index %u opData index %u size %u", index, opDataQueIt->front().index,
778 : opDataQueIt->size());
779 2 : if (index == opDataQueIt->front().index) {
780 2 : opDataInfo = opDataQueIt->front();
781 2 : opDataFound = true; // 需要匹配最后下发的task，不能break
782 : }
783 2 : opDataQueIt->pop();
784 : }
785 2 : if (!opDataFound) {
786 0 : return false;
787 : }
788 :
789 2 : PrintOpDataInfo(opDataInfo, isFftsPlus, stageErrInfo);
790 2 : return true;
791 12 : }
792 :
793 12 : bool TaskExceptionHandler::DealExceptionGroupRank(
794 : rtExceptionInfo* exceptionInfo, std::string& tag, [[maybe_unused]] bool isFftsPlus,
795 : std::string& groupRankContentInfo, std::string& stageErrInfo)
796 : {
797 12 : std::unique_lock<std::mutex> lock(groupRankMapMutex[exceptionInfo->deviceid]);
798 12 : auto groupRankIt = groupRankMap[exceptionInfo->deviceid].find(tag);
799 12 : CHK_PRT_RET(
800 : groupRankIt == groupRankMap[exceptionInfo->deviceid].end(),
801 : HCCL_INFO("tag not found. the fail tag is not from HCCL. tag[%s]", tag.c_str()), false);
802 :
803 6 : auto groupUdiIt = groupUdiMap[exceptionInfo->deviceid].find(groupRankIt->second.first);
804 6 : CHK_PRT_RET(
805 : groupUdiIt == groupUdiMap[exceptionInfo->deviceid].end(),
806 : HCCL_INFO("group not found. the fail group is not from HCCL. group[%s]", groupRankIt->second.first.c_str()),
807 : false);
808 :
809 6 : string peerRankStr;
810 6 : if ((groupRankIt->second.second)->remoteRankId != INVALID_VALUE_RANKSIZE) {
811 0 : peerRankStr += "], peerRankId[";
812 0 : peerRankStr += std::to_string((groupRankIt->second.second)->remoteRankId);
813 : }
814 :
815 6 : string groupRankContent;
816 6 : groupRankContent += "group:[";
817 6 : groupRankContent += groupRankIt->second.first;
818 6 : groupRankContent += "], user define information[";
819 6 : groupRankContent += groupUdiIt->second;
820 6 : groupRankContent += "], rankSize[";
821 6 : groupRankContent += std::to_string((groupRankIt->second.second)->rankSize);
822 6 : groupRankContent += "], rankId[";
823 6 : groupRankContent += std::to_string((groupRankIt->second.second)->rankId);
824 6 : groupRankContent += peerRankStr;
825 6 : groupRankContent += "]";
826 6 : groupRankContentInfo = groupRankContent;
827 :
828 6 : PrintGroupErrorLog(stageErrInfo, groupRankContent, tag);
829 6 : return true;
830 12 : }
831 :
832 5 : bool TaskExceptionHandler::DealExceptionCtx(rtExceptionInfo* exceptionInfo)
833 : {
834 5 : std::unique_lock<std::mutex> lock(opCtxInfoMutex[exceptionInfo->deviceid]);
835 5 : if (!FindAndValidateContext(exceptionInfo)) {
836 2 : return false;
837 : }
838 :
839 3 : FFTSOpInfo fftsOpInfo;
840 3 : CtxInfo exceptionCtxInfo;
841 3 : std::string stageErrInfo = "";
842 :
843 3 : if (!ProcessContext(exceptionInfo, stageErrInfo, fftsOpInfo, exceptionCtxInfo)) {
844 0 : return false;
845 : }
846 :
847 3 : u32 index = fftsOpInfo.index;
848 6 : std::string groupRankContentInfo = "";
849 3 : std::string tag(fftsOpInfo.tag.get());
850 :
851 3 : DealExceptionGroupRank(exceptionInfo, tag, true, groupRankContentInfo, stageErrInfo);
852 3 : DealExceptionOpData(exceptionInfo, tag, true, index, stageErrInfo);
853 3 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, tag);
854 3 : if (!errMsgFlag_.exchange(true)) {
855 2 : if (exceptionCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT) {
856 17 : RPT_INPUT_ERR(
857 : true, "EI0002",
858 : std::vector<std::string>(
859 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
860 : std::vector<std::string>(
861 : {std::to_string(exceptionCtxInfo.GetCtxRemoteUserRank()),
862 : exceptionCtxInfo.GetCtxBaseInfoStr().c_str(), (exceptionCtxInfo.GetCtxParaInfoStr()).c_str(),
863 : groupRankContentInfo.c_str()}));
864 1 : } else if (
865 1 : exceptionCtxInfo.taskType == TaskType::TASK_SDMA
866 1 : || exceptionCtxInfo.taskType == TaskType::TASK_REDUCE_INLINE) {
867 0 : RPT_INPUT_ERR(
868 : true, "EI0012",
869 : std::vector<std::string>(
870 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
871 : std::vector<std::string>(
872 : {std::to_string(exceptionCtxInfo.GetCtxRemoteUserRank()),
873 : exceptionCtxInfo.GetCtxBaseInfoStr().c_str(), (exceptionCtxInfo.GetCtxParaInfoStr()).c_str(),
874 : groupRankContentInfo.c_str()}));
875 : }
876 : }
877 3 : return true;
878 7 : }
879 :
880 5 : bool TaskExceptionHandler::FindAndValidateContext(rtExceptionInfo* exceptionInfo)
881 : {
882 5 : auto mapIt = opCtxInfo[exceptionInfo->deviceid].find(exceptionInfo->streamid);
883 5 : if (mapIt == opCtxInfo[exceptionInfo->deviceid].end()) {
884 0 : HCCL_RUN_INFO("stream not found. the fail ctx is not from HCCL. streamid[%u]", exceptionInfo->streamid);
885 0 : return false;
886 : }
887 :
888 5 : auto& queIt = mapIt->second;
889 5 : if (queIt->size() == 0) {
890 2 : HCCL_ERROR("[TaskExceptionHandler][Callback] CtxOpInfo queue size 0");
891 2 : return false;
892 : }
893 :
894 3 : if ((*(queIt->front().second)).size() == 0) {
895 0 : HCCL_ERROR("[TaskExceptionHandler][Callback] CtxInfoVector size 0");
896 0 : return false;
897 : }
898 :
899 3 : return true;
900 : }
901 :
902 3 : void TaskExceptionHandler::PrintFftsCtxInfo(FFTSOpInfo& fftsOpInfo)
903 : {
904 : // 按照每个task占用128字节打印ffts的子图信息
905 3 : if (fftsOpInfo.descBuf != nullptr && fftsOpInfo.descBufLen > 0) {
906 1 : HCCL_ERROR(
907 : "==========FftsPlusTask-begin-context, ctx_addr=%p, descBuflen=%u, ctx_num=%lu==========",
908 : fftsOpInfo.descBuf.get(), fftsOpInfo.descBufLen, fftsOpInfo.descBufLen / 128UL);
909 2 : for (uint32_t i = 0U; i < (fftsOpInfo.descBufLen / 128UL); i++) {
910 1 : HCCL_ERROR(
911 : "stream_id=%u, task_id=%u, FftsPlusTask context_id=%u:", fftsOpInfo.streamID, fftsOpInfo.taskID, i);
912 1 : uint32_t* buf = reinterpret_cast<uint32_t*>(fftsOpInfo.descBuf.get()) + (i * 32U);
913 5 : for (uint32_t j = 0U; j < 32U; j += 8) {
914 4 : HCCL_ERROR(
915 : "context_id=%u, buf[%02u-%02u]=%08x %08x %08x %08x %08x %08x %08x %08x.", i, j, (j + 7U), buf[j],
916 : buf[j + 1U], buf[j + 2U], buf[j + 3U], buf[j + 4U], buf[j + 5U], buf[j + 6U], buf[j + 7U]);
917 : }
918 : }
919 1 : HCCL_ERROR("==========FftsPlusTask-end-context==========");
920 : }
921 3 : return;
922 : }
923 :
924 3 : bool TaskExceptionHandler::ProcessContext(
925 : rtExceptionInfo* exceptionInfo, std::string& stageErrInfo, FFTSOpInfo& fftsOpInfo, CtxInfo& exceptionCtxInfo)
926 : {
927 3 : auto mapIt = opCtxInfo[exceptionInfo->deviceid].find(exceptionInfo->streamid);
928 3 : auto& queIt = mapIt->second;
929 3 : fftsOpInfo = *(queIt->front().first);
930 3 : exceptionCtxInfo = (*(queIt->front().second))[0];
931 3 : uint16_t invalidCtxid = 65535;
932 3 : bool ctxFound = false;
933 :
934 3 : while (queIt->size() > 0) {
935 3 : if (exceptionInfo->taskid == queIt->back().first->taskID) {
936 3 : fftsOpInfo = *(queIt->back().first);
937 3 : if (exceptionInfo->expandInfo.u.fftsPlusInfo.contextId == invalidCtxid) {
938 : // 子图任务粒度下，RTS返回的异常task不包含contexId时的处理，约定contextId为65535。只记录算子信息
939 0 : HCCL_WARNING(
940 : "%sTask run failed, invalid contexid,"
941 : "base opInformation is %s",
942 : stageErrInfo.c_str(), fftsOpInfo.GetBaseInfoStr().c_str());
943 3 : } else if (exceptionInfo->expandInfo.u.fftsPlusInfo.contextId >= queIt->back().second->size()) {
944 0 : HCCL_ERROR(
945 : "%sTask run failed, contextId[%u] is out of vector "
946 : "size[%zu], base opInformation is %s",
947 : stageErrInfo.c_str(), exceptionInfo->expandInfo.u.fftsPlusInfo.contextId,
948 : queIt->back().second->size(), fftsOpInfo.GetBaseInfoStr().c_str());
949 : } else {
950 3 : exceptionCtxInfo = (*(queIt->back().second))[exceptionInfo->expandInfo.u.fftsPlusInfo.contextId];
951 3 : ctxFound = true;
952 : }
953 3 : break;
954 : } else {
955 0 : queIt->pop_back();
956 : }
957 : }
958 :
959 : auto logKeywordL2
960 3 : = exceptionCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
961 3 : stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + LOG_KEYWORDS_HOST + "]";
962 :
963 3 : PrintFftsCtxInfo(fftsOpInfo);
964 :
965 3 : if (!ctxFound) {
966 0 : return false;
967 : }
968 :
969 3 : if (exceptionCtxInfo.taskType == TaskType::TASK_NOTIFY_WAIT) { // 只在出错task为NotifyWait时打印前序task序列
970 2 : PrintTaskContextInfo(queIt->back().second, exceptionInfo->expandInfo.u.fftsPlusInfo.contextId, stageErrInfo);
971 : }
972 :
973 3 : queIt->clear();
974 :
975 3 : PrintBaseErrorLog(stageErrInfo, fftsOpInfo.GetBaseInfoStr());
976 3 : PrintContextErrorLog(stageErrInfo, exceptionCtxInfo.GetCtxBaseInfoStr());
977 3 : PrintParaErrorLog(stageErrInfo, exceptionCtxInfo.GetCtxParaInfoStr(), std::string(fftsOpInfo.tag.get()));
978 :
979 3 : return true;
980 3 : }
981 :
982 4 : bool TaskExceptionHandler::DealExceptionOp(rtExceptionInfo* exceptionInfo)
983 : {
984 4 : std::unique_lock<std::mutex> lock(opMapMutex[exceptionInfo->deviceid]);
985 4 : bool taskFound = false;
986 4 : auto mapIt = opMap[exceptionInfo->deviceid].find(exceptionInfo->streamid);
987 4 : CHK_PRT_RET(
988 : mapIt == opMap[exceptionInfo->deviceid].end(),
989 : HCCL_RUN_INFO("stream not found. the fail op is not from HCCL. streamid[%u]", exceptionInfo->streamid), false);
990 3 : auto& queIt = mapIt->second;
991 3 : CHK_PRT_RET(queIt->size() == 0, HCCL_ERROR("[TaskExceptionHandler][Callback] OpInfo queue size 0"), false);
992 3 : auto exceptionOpInfo = queIt->back();
993 3 : while (queIt->size() > 0) {
994 3 : if (exceptionInfo->taskid == queIt->back().taskID) {
995 3 : exceptionOpInfo = queIt->back();
996 3 : taskFound = true; // 从后往前匹配最后下发的相同taskId
997 3 : break;
998 : }
999 0 : queIt->pop_back();
1000 : }
1001 3 : if (!taskFound) {
1002 0 : return false;
1003 : }
1004 3 : queIt->clear();
1005 :
1006 : auto logKeywordL2
1007 3 : = exceptionInfo->retcode == ACL_ERROR_RT_FFTS_PLUS_TIMEOUT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
1008 3 : auto stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + LOG_KEYWORDS_HOST + "]";
1009 :
1010 3 : PrintBaseErrorLog(stageErrInfo, exceptionOpInfo.GetBaseInfoStr());
1011 3 : u32 index = exceptionOpInfo.index;
1012 6 : std::string groupRankContentInfo = "";
1013 3 : std::string tag(exceptionOpInfo.tag.get());
1014 3 : DealExceptionGroupRank(exceptionInfo, tag, true, groupRankContentInfo, stageErrInfo);
1015 3 : DealExceptionOpData(exceptionInfo, tag, true, index, stageErrInfo);
1016 3 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, tag);
1017 3 : if (!errMsgFlag_.exchange(true)) {
1018 2 : if (exceptionInfo->retcode == ACL_ERROR_RT_FFTS_PLUS_TIMEOUT) {
1019 16 : RPT_INPUT_ERR(
1020 : true, "EI0002",
1021 : std::vector<std::string>(
1022 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1023 : std::vector<std::string>(
1024 : {"unknown", exceptionOpInfo.GetBaseInfoStr().c_str(), errMsg.c_str(),
1025 : groupRankContentInfo.c_str()}));
1026 : }
1027 : }
1028 3 : return true;
1029 6 : }
1030 :
1031 3 : void TaskExceptionHandler::PrintTaskContextInfo(
1032 : const std::shared_ptr<std::deque<TaskInfo>>& taskQue, std::string& stageErrInfo)
1033 : {
1034 3 : HCCL_ERROR(
1035 : "%sTask run failed, context sequence before error task is "
1036 : "[NotifyRecord:NR(rank,id), NotifyWait:NW(rank,id), Memcpy:M(rank), Reduce: R(rank), "
1037 : "InlineReduce:IR(rank), RDMASend:RS(rank,id)]:",
1038 : stageErrInfo.c_str());
1039 3 : std::string taskContextInfo = "";
1040 3 : u32 startIndex = (taskQue->size() > TASK_CONTEXT_SIZE) ? (taskQue->size() - TASK_CONTEXT_SIZE) : 0;
1041 3 : for (; startIndex < taskQue->size(); startIndex++) {
1042 0 : auto taskInfo = taskQue->at(startIndex);
1043 :
1044 0 : std::string taskStr = GetTaskBriefsName(taskInfo.taskType);
1045 0 : taskStr += "(";
1046 0 : taskStr += taskInfo.GetRankInfo();
1047 0 : if (taskInfo.taskType == TaskType::TASK_NOTIFY_RECORD || taskInfo.taskType == TaskType::TASK_NOTIFY_WAIT
1048 0 : || taskInfo.taskType == TaskType::TASK_RDMA) {
1049 0 : taskStr += ("," + taskInfo.GetNotifyInfo());
1050 : }
1051 0 : taskStr += "),";
1052 0 : if (taskContextInfo.size() + taskStr.size() >= TASK_CONTEXT_INFO_SIZE) {
1053 0 : HCCL_ERROR("%s%s ...", stageErrInfo.c_str(), taskContextInfo.c_str());
1054 0 : taskContextInfo = "";
1055 : }
1056 0 : taskContextInfo += taskStr;
1057 0 : }
1058 3 : HCCL_ERROR("%s%s end.", stageErrInfo.c_str(), taskContextInfo.c_str());
1059 6 : return;
1060 3 : }
1061 :
1062 0 : void TaskExceptionHandler::ParseTaskSyncFlag(s32* flagMem, u32 flagMemSize, u32 rankSize, u32 rank, u32 index)
1063 : {
1064 0 : u32 chips1v1 = std::min(rankSize * NUM_BLOCKS_PER_RANK, MAX_RANK_SIZE_SUPERPOD) * NOTIFY_NUM * INTERVAL_1V1;
1065 0 : u32 cores1v1 = MAX_NUM_BLOCKS * NOTIFY_GROUPS_1V1 * INTERVAL_1V1;
1066 0 : u32 chips1vN = PRINT_1VN_NUM * INTERVAL_1VN * NOTIFY_GROUPS_1V1;
1067 0 : u32 cores1vN = PRINT_1VN_NUM * INTERVAL_1VN * NOTIFY_GROUPS_1V1;
1068 0 : u32 chipsNv1 = PRINT_NV1_NUM * INTERVAL_NV1 * NOTIFY_GROUPS_1V1;
1069 0 : u32 coresNv1 = PRINT_NV1_NUM * INTERVAL_NV1 * NOTIFY_GROUPS_1V1;
1070 0 : u32 count = rankSize * CORE_PER_CARDS * INTERVAL_COUNT;
1071 0 : u32 syncCount = (chips1v1 + cores1v1 + chips1vN + cores1vN + chipsNv1 + coresNv1) * PING_PONG_NUM + count;
1072 0 : u32 total = syncCount * sizeof(u32);
1073 0 : if (total > flagMemSize) {
1074 0 : HCCL_ERROR("rank %u opIndex=%u flag mem size %u is too little total %u.", rank, index, flagMemSize, total);
1075 0 : return;
1076 : }
1077 :
1078 0 : s32* buf = flagMem;
1079 0 : u32 offset = 0;
1080 :
1081 0 : const std::string PREFIX[PING_PONG_NUM] = {"ping", "pong"};
1082 0 : std::string str;
1083 0 : for (u32 i = 0; i < PING_PONG_NUM; ++i) {
1084 : // print chips1v1
1085 0 : str = SerializeSyncFlag(buf + offset, rankSize * NUM_BLOCKS_PER_RANK * NOTIFY_NUM, INTERVAL_1V1);
1086 0 : offset += chips1v1;
1087 0 : HCCL_ERROR("rank %u opIndex %u chips 1v1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1088 :
1089 0 : str = SerializeSyncFlag(buf + offset, MAX_NUM_BLOCKS * NOTIFY_GROUPS_1V1, INTERVAL_1V1);
1090 0 : offset += cores1v1;
1091 0 : HCCL_ERROR("rank %u opIndex %u cores 1v1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1092 :
1093 0 : str = SerializeSyncFlag(buf + offset, PRINT_1VN_NUM * NOTIFY_GROUPS_1V1, INTERVAL_1VN);
1094 0 : offset += chips1vN;
1095 0 : HCCL_ERROR("rank %u opIndex %u chips 1vn sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1096 :
1097 0 : str = SerializeSyncFlag(buf + offset, PRINT_1VN_NUM * NOTIFY_GROUPS_1V1, INTERVAL_1VN);
1098 0 : offset += cores1vN;
1099 0 : HCCL_ERROR("rank %u opIndex %u cores 1vn sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1100 :
1101 0 : str = SerializeSyncFlag(buf + offset, PRINT_NV1_NUM * NOTIFY_GROUPS_1V1, INTERVAL_NV1);
1102 0 : offset += chipsNv1;
1103 0 : HCCL_ERROR("rank %u opIndex %u chips nv1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1104 :
1105 0 : str = SerializeSyncFlag(buf + offset, PRINT_NV1_NUM * NOTIFY_GROUPS_1V1, INTERVAL_NV1);
1106 0 : offset += coresNv1;
1107 0 : HCCL_ERROR("rank %u opIndex %u cores nv1 sync flag [%s] %s", rank, index, PREFIX[i].c_str(), str.c_str());
1108 : }
1109 0 : str = SerializeSyncFlag(buf + offset, rankSize * CORE_PER_CARDS, INTERVAL_COUNT);
1110 0 : HCCL_ERROR("rank %u opIndex %u sync count [%s]", rank, index, str.c_str());
1111 0 : }
1112 :
1113 0 : std::string TaskExceptionHandler::SerializeSyncFlag(s32* buf, u32 num, u32 interval)
1114 : {
1115 0 : std::stringstream ss;
1116 0 : s32* pos = buf;
1117 0 : for (u32 i = 0; i < num; i = i + 1) {
1118 0 : ss << std::dec << " " << *pos;
1119 0 : pos = pos + interval;
1120 : }
1121 0 : return ss.str();
1122 0 : }
1123 :
1124 0 : void TaskExceptionHandler::PrintTaskAivBuffer(const std::shared_ptr<std::deque<TaskInfo>>& taskQue)
1125 : {
1126 0 : if (taskQue->empty()) {
1127 0 : return;
1128 : }
1129 : // width参考aiv_communication_base.cc的MAX_FLAG_SIZE_PER_KERNEL
1130 :
1131 0 : u32 flagMemSize = 1024 * 1024;
1132 0 : auto& taskInfo = taskQue->back();
1133 0 : u32 realRankSize = taskInfo.taskPara.Aiv.rankSize;
1134 0 : void* tmpFlagMem = malloc(flagMemSize);
1135 0 : if (tmpFlagMem == nullptr) {
1136 0 : return;
1137 : }
1138 0 : s32* flagMem = static_cast<s32*>(tmpFlagMem);
1139 0 : hrtMemSyncCopy(
1140 0 : flagMem, flagMemSize, reinterpret_cast<u8*>(taskInfo.taskPara.Aiv.flagMem), flagMemSize,
1141 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_HOST);
1142 :
1143 0 : ParseTaskSyncFlag(flagMem, flagMemSize, realRankSize, taskInfo.taskPara.Aiv.rank, taskInfo.index);
1144 0 : free(flagMem);
1145 : }
1146 :
1147 0 : void TaskExceptionHandler::PrintTaskAivInfo(const std::shared_ptr<std::deque<TaskInfo>>& taskQue)
1148 : {
1149 0 : HCCL_ERROR("[PrintTaskAivInfo] print start: ");
1150 : // 从后往前遍历，最多打印PRINT_TASK_AIV_INFO_COUNT个taskAiv
1151 0 : int cnt = PRINT_TASK_AIV_INFO_COUNT;
1152 0 : for (auto it = taskQue->end() - 1; it >= taskQue->begin(); --it) {
1153 0 : if (!it->isAlgInfo) {
1154 0 : continue;
1155 : }
1156 0 : if (cnt <= 0) {
1157 0 : break;
1158 : }
1159 0 : auto taskInfo = *it;
1160 0 : HCCL_ERROR("[AIV](%s) ", taskInfo.GetParaAiv().c_str());
1161 0 : cnt--;
1162 0 : }
1163 0 : HCCL_ERROR("[PrintTaskAivInfo] print end.");
1164 0 : return;
1165 : }
1166 :
1167 5 : void splitAndPrintErrStr(const std::string& s)
1168 : {
1169 5 : std::vector<string> parts;
1170 5 : std::istringstream iss(s);
1171 5 : std::string part;
1172 :
1173 : // 将字符串按照空格分隔
1174 114 : while (iss >> part) {
1175 109 : parts.push_back(part);
1176 : }
1177 :
1178 : // 每10组作为一行打印，暂不做通用化处理
1179 5 : constexpr u32 plen = 10;
1180 19 : for (size_t i = 0; i < parts.size(); i += plen) {
1181 14 : std::string line;
1182 123 : for (size_t j = i; j < i + plen && j < parts.size(); ++j) {
1183 109 : if (j != i) {
1184 95 : line += " ";
1185 : }
1186 109 : line += parts[j];
1187 : }
1188 14 : HCCL_ERROR("%s", line.c_str());
1189 14 : }
1190 5 : }
1191 :
1192 11 : HcclResult TaskExceptionHandler::PrintCommAivInfo()
1193 : {
1194 : /* 本函数的目的：在任务失败后，遍历当前device的所有通信域
1195 : 对于通信域内存在AIV算子的情况进行统计和打印
1196 : 提示用户如果有多个通信域存在AIV算子可能导致执行卡住
1197 : */
1198 11 : u32 groupHasAivCount = 0;
1199 11 : u32 groupNoAivCount = 0;
1200 11 : s32 deviceLogicId = -1;
1201 11 : std::stringstream groupHasAivInfo;
1202 11 : std::stringstream groupNoAivInfo;
1203 :
1204 11 : HcclResult ret = hrtGetDevice(&deviceLogicId);
1205 11 : if (ret != HCCL_SUCCESS) {
1206 1 : HCCL_ERROR("[TaskExceptionHandler][PrintCommAivInfo]hrtGetDevice failed, ret[%d]", ret);
1207 1 : return HCCL_E_PARA;
1208 : }
1209 :
1210 : // 轮询aivGroupIndexMap_[deviceLogicId]的group，确认是否此group内有aiv算子
1211 : // 对于存在aiv算子的，记录和打印group信息和aiv信息
1212 10 : if (aivGroupIndexMap_[deviceLogicId].size() == 0) {
1213 6 : HCCL_ERROR("[TaskExceptionHandler][PrintCommAivInfo] aiv group not record");
1214 6 : return HCCL_SUCCESS;
1215 : }
1216 :
1217 113 : for (auto it = aivGroupIndexMap_[deviceLogicId].begin(); it != aivGroupIndexMap_[deviceLogicId].end(); it++) {
1218 109 : if (it->second == 0) {
1219 5 : groupNoAivInfo << "[" << it->first.c_str() << "] ";
1220 5 : groupNoAivCount++;
1221 : } else {
1222 104 : groupHasAivInfo << "[" << it->first.c_str() << "] ";
1223 104 : groupHasAivCount++;
1224 : }
1225 : }
1226 :
1227 : // 如果遍历发现，存在通信域内执行过aiv算子，则提示有可能有卡死风险；大于0则提示，因为MC2也有可能有aiv算子。
1228 4 : if (groupHasAivCount != 0) {
1229 3 : HCCL_ERROR(
1230 : "[TaskExceptionHandler][PrintCommAivInfo] multi groups include aiv alg, may cause execution stuck."
1231 : " has aiv group count[%u]",
1232 : groupHasAivCount);
1233 3 : HCCL_ERROR("groups has aiv list[groupName]:");
1234 3 : splitAndPrintErrStr(groupHasAivInfo.str());
1235 : }
1236 :
1237 : // 通信域不包含aiv算子的，也一并提示
1238 4 : if (groupNoAivCount != 0) {
1239 2 : HCCL_ERROR("[TaskExceptionHandler][PrintCommAivInfo] no aiv alg group count[%u].", groupNoAivCount);
1240 2 : HCCL_ERROR("groups no aiv list[groupName]: ");
1241 2 : splitAndPrintErrStr(groupNoAivInfo.str());
1242 : }
1243 :
1244 4 : return HCCL_SUCCESS;
1245 11 : }
1246 :
1247 7 : bool TaskExceptionHandler::DealExceptionTask(rtExceptionInfo* exceptionInfo)
1248 : {
1249 7 : std::unique_lock<std::mutex> lock(taskMapMutex[exceptionInfo->deviceid]);
1250 7 : bool taskFound = false;
1251 7 : auto mapIt = taskMap[exceptionInfo->deviceid].find(exceptionInfo->streamid);
1252 7 : CHK_PRT_RET(
1253 : mapIt == taskMap[exceptionInfo->deviceid].end(),
1254 : HCCL_RUN_INFO("stream not found. the fail task is not from HCCL. streamid[%u]", exceptionInfo->streamid),
1255 : false);
1256 6 : auto& queIt = mapIt->second;
1257 6 : CHK_PRT_RET(queIt->size() == 0, HCCL_ERROR("[TaskExceptionHandler][Callback] TaskInfo queue size 0"), false);
1258 :
1259 : // 从后往前匹配最后下发的相同taskId
1260 5 : auto exceptionTaskInfo = queIt->back();
1261 6 : while (queIt->size() > 0) {
1262 6 : if (exceptionInfo->taskid == queIt->back().taskID) {
1263 5 : exceptionTaskInfo = queIt->back();
1264 5 : taskFound = true;
1265 5 : break;
1266 : }
1267 1 : queIt->pop_back();
1268 : }
1269 5 : if (!taskFound) {
1270 0 : return false;
1271 : }
1272 :
1273 : // 检测是否存在多通信域有aiv算子情况，提示可能导致执行卡住
1274 5 : CHK_PRT_RET(PrintCommAivInfo(), HCCL_ERROR("[TaskExceptionHandler] PrintCommAivInfo failed."), false);
1275 :
1276 5 : std::string logKeywordL2;
1277 5 : std::string logKeywordL3;
1278 :
1279 5 : if (exceptionTaskInfo.isAlgInfo) {
1280 : // aiv场景若根据retCode是否为ACL_ERROR_RT_VECTOR_CORE_TIMEOUT判断是否为超时报错
1281 0 : logKeywordL2 = exceptionInfo->retcode == ACL_ERROR_RT_VECTOR_CORE_TIMEOUT ? LOG_KEYWORDS_TIMEOUT :
1282 0 : LOG_KEYWORDS_RUN_FAILED;
1283 0 : logKeywordL3 = LOG_KEYWORDS_AIV;
1284 : } else {
1285 : // 非aiv场景根据当前报错的taskType是否为TASK_NOTIFY_WAIT判断是否为超时报错
1286 : logKeywordL2
1287 5 : = exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
1288 5 : logKeywordL3 = LOG_KEYWORDS_HOST_TS;
1289 : }
1290 :
1291 5 : auto stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + logKeywordL3 + "]";
1292 :
1293 5 : if (exceptionTaskInfo.isAlgInfo) {
1294 0 : PrintTaskAivBuffer(queIt);
1295 0 : PrintTaskAivInfo(queIt);
1296 5 : } else if (exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT) {
1297 3 : queIt->pop_back();
1298 : // 只在出错task为NotifyWait时打印前序task序列
1299 3 : PrintTaskContextInfo(queIt, stageErrInfo);
1300 : }
1301 :
1302 5 : queIt->clear();
1303 5 : HCCL_ERROR("%sTask from HCCL run failed.", stageErrInfo.c_str());
1304 : // 防止tag字符串过长， 信息分开打印
1305 5 : PrintBaseErrorLog(stageErrInfo, exceptionTaskInfo.GetBaseInfoStr());
1306 5 : PrintParaErrorLog(stageErrInfo, exceptionTaskInfo.GetParaInfoStr(), exceptionTaskInfo.tag);
1307 5 : u32 index = exceptionTaskInfo.index;
1308 5 : std::string groupRankContentInfo = "";
1309 5 : if (!exceptionTaskInfo.isAlgInfo) {
1310 : // AlgInfo时不打印group rank等信息
1311 5 : DealExceptionGroupRank(exceptionInfo, exceptionTaskInfo.tag, false, groupRankContentInfo, stageErrInfo);
1312 : }
1313 5 : DealExceptionOpData(exceptionInfo, exceptionTaskInfo.tag, false, index, stageErrInfo);
1314 5 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, exceptionTaskInfo.tag);
1315 5 : if (!errMsgFlag_.exchange(true)) {
1316 4 : if (logKeywordL2 == LOG_KEYWORDS_TIMEOUT) {
1317 34 : RPT_INPUT_ERR(
1318 : true, "EI0002",
1319 : std::vector<std::string>(
1320 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1321 : std::vector<std::string>(
1322 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()), exceptionTaskInfo.GetBaseInfoStr().c_str(),
1323 : (exceptionTaskInfo.GetParaInfoStr()).c_str(), groupRankContentInfo.c_str()}));
1324 : } else {
1325 34 : RPT_INPUT_ERR(
1326 : true, "EI0012",
1327 : std::vector<std::string>(
1328 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1329 : std::vector<std::string>(
1330 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()), exceptionTaskInfo.GetBaseInfoStr().c_str(),
1331 : (exceptionTaskInfo.GetParaInfoStr()).c_str(), groupRankContentInfo.c_str()}));
1332 : }
1333 : }
1334 5 : return true;
1335 15 : }
1336 :
1337 11 : void TaskExceptionHandler::PrintAicpuErrorMessage(rtExceptionInfo* exceptionInfo, bool& isExistAicpuError)
1338 : {
1339 11 : ErrorMessageReport errorMessage;
1340 11 : unique_lock<std::mutex> lock(g_commHadCallbackArrayMutex);
1341 11 : if (g_commHadCallbackArray[exceptionInfo->deviceid]) {
1342 : // 防止同一个device上出现通信主流和kernel流均出现task exception时runtime调用两次callback
1343 : // HDC通道信息不是读清，防止aicpu task exception重复上报
1344 2 : HCCL_WARNING("aicpu error message been reported. deviceid[%u]", exceptionInfo->deviceid);
1345 2 : return;
1346 : }
1347 9 : lock.unlock();
1348 9 : if (g_communicatorCallbackMap[exceptionInfo->deviceid].find(exceptionInfo->streamid)
1349 18 : != g_communicatorCallbackMap[exceptionInfo->deviceid].end()) {
1350 : // 找到对应的通信域，并调用回调函数从HDC通道获取AICPU异常信息
1351 7 : errorMessage = (g_communicatorCallbackMap[exceptionInfo->deviceid])[exceptionInfo->streamid]();
1352 7 : if (strlen(errorMessage.tag) > 0) {
1353 6 : isExistAicpuError = true;
1354 6 : string groupRankContent;
1355 6 : u32 streamId = static_cast<u32>(errorMessage.streamId);
1356 6 : std::string tag = std::string(errorMessage.tag);
1357 6 : u32 index = 0;
1358 : TaskParaNotify para(
1359 6 : static_cast<u64>(errorMessage.notifyId), errorMessage.stage, errorMessage.remoteUserRank);
1360 : TaskInfo exceptionTaskInfo(
1361 6 : streamId, errorMessage.taskId, tag, errorMessage.taskType, errorMessage.algType, index, para);
1362 6 : auto logKeywordL2 = exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT ? LOG_KEYWORDS_TIMEOUT :
1363 6 : LOG_KEYWORDS_RUN_FAILED;
1364 6 : auto stageErrInfo = "[" + LOG_KEYWORDS_TASK_EXEC + "][" + logKeywordL2 + "][" + LOG_KEYWORDS_AICPU + "]";
1365 6 : HCCL_ERROR("%sTask from HCCL run failed.", stageErrInfo.c_str());
1366 : // 防止tag字符串过长， 信息分开打印
1367 6 : PrintBaseErrorLog(stageErrInfo, exceptionTaskInfo.GetBaseInfoStr());
1368 6 : PrintParaErrorLog(stageErrInfo, exceptionTaskInfo.GetParaInfoStr(), exceptionTaskInfo.tag);
1369 6 : PrintGroupErrorMessage(errorMessage, exceptionTaskInfo, groupRankContent, stageErrInfo);
1370 6 : PrintOpDataErrorMessage(exceptionInfo->deviceid, errorMessage, stageErrInfo);
1371 6 : std::string errMsg = GetAndPrintHeartbeatErr(exceptionInfo, tag);
1372 6 : if (!errMsgFlag_.exchange(true)) {
1373 3 : if (exceptionTaskInfo.taskType == TaskType::TASK_NOTIFY_WAIT) {
1374 34 : RPT_INPUT_ERR(
1375 : true, "EI0002",
1376 : std::vector<std::string>(
1377 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1378 : std::vector<std::string>(
1379 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()),
1380 : exceptionTaskInfo.GetBaseInfoStr().c_str(), (exceptionTaskInfo.GetParaInfoStr()).c_str(),
1381 : "none"}));
1382 1 : } else if (
1383 1 : exceptionTaskInfo.taskType == TaskType::TASK_SDMA
1384 0 : || exceptionTaskInfo.taskType == TaskType::TASK_REDUCE_INLINE) {
1385 18 : RPT_INPUT_ERR(
1386 : true, "EI0012",
1387 : std::vector<std::string>(
1388 : {"remote_rankid", "base_information", "task_information", "group_rank_content"}),
1389 : std::vector<std::string>(
1390 : {std::to_string(exceptionTaskInfo.GetRemoteUserRank()),
1391 : exceptionTaskInfo.GetBaseInfoStr().c_str(),
1392 : (exceptionTaskInfo.GetParaInfoStr() + errMsg).c_str(), groupRankContent.c_str()}));
1393 : }
1394 : }
1395 6 : lock.lock();
1396 6 : g_commHadCallbackArray[exceptionInfo->deviceid] = true;
1397 6 : }
1398 : } else {
1399 2 : HCCL_INFO("PrintAicpuErrorMessage streamId[%d] is not found.", exceptionInfo->streamid);
1400 : }
1401 9 : return;
1402 18 : }
1403 :
1404 6 : void TaskExceptionHandler::PrintGroupErrorMessage(
1405 : ErrorMessageReport& errorMessage, TaskInfo& exceptionTaskInfo, string& groupRankContent, string& stageErrInfo)
1406 : {
1407 6 : std::string groupUdi;
1408 6 : std::string groupName = std::string(errorMessage.group);
1409 6 : ProfilerBase::GetUdiByGroup(groupName, groupUdi);
1410 :
1411 6 : groupRankContent += "group:[";
1412 6 : groupRankContent += std::string(errorMessage.group);
1413 6 : groupRankContent += "], user define information[";
1414 6 : groupRankContent += groupUdi;
1415 6 : groupRankContent += "], rankSize[";
1416 6 : groupRankContent += std::to_string(errorMessage.rankSize);
1417 6 : groupRankContent += "], rankId[";
1418 6 : groupRankContent += std::to_string(errorMessage.rankId);
1419 6 : groupRankContent += " ";
1420 6 : groupRankContent += std::to_string(errorMessage.remoteUserRank);
1421 6 : groupRankContent += "]";
1422 :
1423 6 : PrintGroupErrorLog(stageErrInfo, groupRankContent, exceptionTaskInfo.tag);
1424 12 : return;
1425 6 : }
1426 :
1427 6 : void TaskExceptionHandler::PrintOpDataErrorMessage(u32 deviceId, ErrorMessageReport& errorMessage, string& stageErrInfo)
1428 : {
1429 6 : stringstream opDataStr;
1430 6 : opDataStr << "src" << "[0x" << std::hex << errorMessage.srcAddr << "], dst[0x" << std::hex << errorMessage.dstAddr
1431 6 : << "], ";
1432 :
1433 6 : string opStr;
1434 6 : if (errorMessage.reduceType != HcclReduceOp::HCCL_REDUCE_RESERVED) {
1435 0 : opStr += "reduceType[";
1436 0 : opStr += GetReduceOpEnumStr(static_cast<HcclReduceOp>(errorMessage.reduceType));
1437 0 : opStr += "], ";
1438 : }
1439 :
1440 6 : string opDataContent;
1441 6 : opDataContent += "deviceId:[";
1442 6 : opDataContent += std::to_string(deviceId);
1443 6 : opDataContent += "], index[";
1444 6 : opDataContent += std::to_string(errorMessage.opIndex);
1445 6 : opDataContent += "], count[";
1446 6 : opDataContent += std::to_string(errorMessage.count);
1447 6 : opDataContent += "], ";
1448 6 : opDataContent += opStr;
1449 6 : opDataContent += opDataStr.str();
1450 6 : opDataContent += "dataType[";
1451 6 : opDataContent += GetDataTypeEnumStr(errorMessage.dataType);
1452 6 : opDataContent += "].";
1453 :
1454 6 : PrintOpDataErrorLog(stageErrInfo, opDataContent);
1455 12 : return;
1456 6 : }
1457 :
1458 2 : void TaskExceptionHandler::Callback(rtExceptionInfo* exceptionInfo)
1459 : {
1460 2 : HCCL_RUN_INFO("[TaskExceptionHandler][%s]begin to execute hccl task exception callback function.", __func__);
1461 2 : bool isExistAicpuError = false;
1462 2 : if (exceptionInfo == nullptr) {
1463 0 : HCCL_ERROR("[TaskExceptionHandler][Callback] exceptionInfo is nullptr.");
1464 0 : return;
1465 : }
1466 :
1467 2 : PrintAicpuErrorMessage(exceptionInfo, isExistAicpuError);
1468 2 : if (isExistAicpuError) {
1469 : // 如果已经有AICPU上报的task exception, 则host侧无需再次重复上报
1470 0 : return;
1471 : }
1472 : u32 maxDeviceNum;
1473 2 : HcclResult ret = GetMaxDevNum(maxDeviceNum);
1474 2 : if (ret != HCCL_SUCCESS) {
1475 0 : HCCL_ERROR("[GetMaxDevNum] get maxDeviceNum error");
1476 0 : return;
1477 : }
1478 2 : CHK_PRT_RET(
1479 : exceptionInfo->deviceid >= maxDeviceNum,
1480 : HCCL_WARNING(
1481 : "deviceID[%u] from exceptionInfo is bigger than maxDeviceNum[%u]", exceptionInfo->deviceid,
1482 : maxDeviceNum), );
1483 2 : SaluSleep(ONE_MILLISECOND_OF_USLEEP); // sleep 1ms，等待task被存入数据结构
1484 2 : HCCL_DEBUG(
1485 : "[TaskExceptionHandler][Callback]Task run failed, ffts+ task type:%d, TaskExceptionSwitch:%u",
1486 : exceptionInfo->expandInfo.type, GetExternalInputTaskExceptionSwitch());
1487 2 : if (exceptionInfo->expandInfo.type == RT_EXCEPTION_FFTS_PLUS) {
1488 1 : if (GetExternalInputTaskExceptionSwitch() == 1) {
1489 1 : DealExceptionCtx(exceptionInfo); // 子任务粒度
1490 : } else {
1491 0 : DealExceptionOp(exceptionInfo); // 算子粒度
1492 : }
1493 : } else {
1494 1 : DealExceptionTask(exceptionInfo);
1495 : }
1496 2 : return;
1497 : }
1498 506 : HcclResult TaskExceptionHandler::Init()
1499 : {
1500 506 : if (communicatorCount_.fetch_add(1) == 0) {
1501 182 : HCCL_RUN_INFO("[TaskExceptionHandler][%s] register taskFailCallback", __func__);
1502 182 : CHK_RET(hrtRegTaskFailCallbackByModule(Callback));
1503 182 : CHK_RET(hrtGetStreamAvailableNum(maxStrCount));
1504 182 : maxStrCount = (maxStrCount < STREAM_COUNT_UPPER_LIMIT) ? maxStrCount : STREAM_COUNT_UPPER_LIMIT;
1505 : }
1506 506 : maxTaskCount = TASK_COUNT_UPPER_LIMIT;
1507 : // 单算子模式task过多的特殊处理
1508 506 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
1509 449 : maxTaskCount = TASK_COUNT_UPPER_LIMIT_OP_BASE;
1510 : }
1511 :
1512 506 : HCCL_INFO("get from RTS the max stream count[%u] the max task count[%u]", maxStrCount, maxTaskCount);
1513 :
1514 621 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1515 621 : && GetExternalInputTaskExceptionSwitch() == 1) {
1516 330 : for (std::vector<CtxInfo>& ctxInfoVector : ctxInfoArray) {
1517 325 : ctxInfoVector.reserve(100); // vector预留100个ctxInfo空间
1518 : }
1519 : }
1520 :
1521 : // 对全局变量g_commHadCallbackArray进行初始化
1522 33396 : for (u32 i = 0; i < MAX_MODULE_DEVICE_NUM; i++) {
1523 32890 : g_commHadCallbackArray[i] = false;
1524 : }
1525 506 : return HCCL_SUCCESS;
1526 : }
1527 :
1528 808 : HcclResult TaskExceptionHandler::DeInit()
1529 : {
1530 808 : if (communicatorCount_.fetch_sub(1) == 1) {
1531 181 : CHK_RET(hrtRegTaskFailCallbackByModule(nullptr));
1532 181 : HCCL_RUN_INFO("deInit taskFailCallback");
1533 : }
1534 808 : return HCCL_SUCCESS;
1535 : }
1536 :
1537 7 : bool IsOneSideTask(u32 streamId)
1538 : {
1539 7 : std::string tag;
1540 7 : CHK_PRT(ProfilerBase::GetTagByStream(streamId, tag));
1541 7 : if (tag.find("BatchPut_") != std::string::npos || tag.find("BatchGet_") != std::string::npos) {
1542 0 : return true;
1543 : }
1544 7 : return false;
1545 7 : }
1546 :
1547 2 : HcclResult TaskExceptionHandler::Save(
1548 : u32 captureStreamID, u32 streamID, u32 taskID, TaskType& taskType, const TaskParaNotify& para)
1549 : {
1550 : u32 maxDeviceNum;
1551 2 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1552 2 : CHK_PRT_RET(
1553 : deviceLogicId_ >= maxDeviceNum,
1554 : HCCL_ERROR(
1555 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1556 : maxDeviceNum),
1557 : HCCL_E_INTERNAL);
1558 2 : HCCL_INFO(
1559 : "[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u], taskType[%d]", __func__, streamID, taskID,
1560 : taskType);
1561 4 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1562 4 : && GetExternalInputTaskExceptionSwitch() == 1 && !IsOneSideTask(captureStreamID)) {
1563 2 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1564 2 : CtxInfo tmpCtxInfo(taskType, para);
1565 2 : ctxInfoArray[deviceLogicId_].insert(ctxInfoArray[deviceLogicId_].end(), tmpCtxInfo);
1566 2 : return HCCL_SUCCESS;
1567 2 : }
1568 :
1569 0 : std::string tag;
1570 0 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1571 0 : AlgType algType = AlgType::Reserved();
1572 0 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1573 0 : u32 index = 0;
1574 0 : ProfilerBase::GetSubmittedOpCnt(index);
1575 :
1576 0 : TaskInfo tmpTaskInfo(streamID, taskID, tag, taskType, algType, index, para);
1577 0 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1578 :
1579 0 : CHK_RET(InsertRankInfo(tag));
1580 0 : CHK_RET(InsertOpData(tag));
1581 0 : return HCCL_SUCCESS;
1582 0 : }
1583 :
1584 2 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, TaskType& taskType, const TaskParaNotify& para)
1585 : {
1586 2 : return Save(streamID, streamID, taskID, taskType, para);
1587 : }
1588 :
1589 : HcclResult
1590 37 : TaskExceptionHandler::Save(u32 captureStreamID, u32 streamID, u32 taskID, TaskType& taskType, const TaskParaDMA& para)
1591 : {
1592 : u32 maxDeviceNum;
1593 37 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1594 37 : CHK_PRT_RET(
1595 : deviceLogicId_ >= maxDeviceNum,
1596 : HCCL_ERROR(
1597 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1598 : maxDeviceNum),
1599 : HCCL_E_INTERNAL);
1600 37 : HCCL_INFO(
1601 : "[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u], taskType[%d]", __func__, streamID, taskID,
1602 : taskType);
1603 46 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1604 46 : && GetExternalInputTaskExceptionSwitch() == 1 && !IsOneSideTask(captureStreamID)) {
1605 5 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1606 5 : CtxInfo tmpCtxInfo(taskType, para);
1607 5 : ctxInfoArray[deviceLogicId_].insert(ctxInfoArray[deviceLogicId_].end(), tmpCtxInfo);
1608 5 : return HCCL_SUCCESS;
1609 5 : }
1610 :
1611 32 : std::string tag;
1612 32 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1613 32 : AlgType algType = AlgType::Reserved();
1614 32 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1615 32 : u32 index = 0;
1616 32 : ProfilerBase::GetSubmittedOpCnt(index);
1617 :
1618 32 : TaskInfo tmpTaskInfo(streamID, taskID, tag, taskType, algType, index, para);
1619 32 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1620 0 : CHK_RET(InsertRankInfo(tag));
1621 0 : CHK_RET(InsertOpData(tag));
1622 0 : return HCCL_SUCCESS;
1623 32 : }
1624 :
1625 37 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, TaskType& taskType, const TaskParaDMA& para)
1626 : {
1627 37 : return Save(streamID, streamID, taskID, taskType, para);
1628 : }
1629 :
1630 9 : HcclResult TaskExceptionHandler::Save(
1631 : u32 captureStreamID, u32 streamID, u32 taskID, TaskType& taskType, const TaskParaReduce& para)
1632 : {
1633 : u32 maxDeviceNum;
1634 9 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1635 9 : CHK_PRT_RET(
1636 : deviceLogicId_ >= maxDeviceNum,
1637 : HCCL_ERROR(
1638 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1639 : maxDeviceNum),
1640 : HCCL_E_INTERNAL);
1641 9 : HCCL_INFO(
1642 : "[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u], taskType[%d]", __func__, streamID, taskID,
1643 : taskType);
1644 9 : if (GetExternalInputHcclEnableFfts() && GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1645 9 : && GetExternalInputTaskExceptionSwitch() == 1 && !IsOneSideTask(captureStreamID)) {
1646 0 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1647 0 : CtxInfo tmpCtxInfo(taskType, para);
1648 0 : ctxInfoArray[deviceLogicId_].insert(ctxInfoArray[deviceLogicId_].end(), tmpCtxInfo);
1649 0 : return HCCL_SUCCESS;
1650 0 : }
1651 :
1652 9 : std::string tag;
1653 9 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1654 9 : AlgType algType = AlgType::Reserved();
1655 9 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1656 9 : u32 index = 0;
1657 9 : ProfilerBase::GetSubmittedOpCnt(index);
1658 :
1659 9 : TaskInfo tmpTaskInfo(streamID, taskID, tag, taskType, algType, index, para);
1660 9 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1661 0 : CHK_RET(InsertRankInfo(tag));
1662 0 : CHK_RET(InsertOpData(tag));
1663 0 : return HCCL_SUCCESS;
1664 9 : }
1665 :
1666 5 : HcclResult TaskExceptionHandler::Save(u32 captureStreamID, u32 streamID, u32 taskID, const TaskParaAiv& para)
1667 : {
1668 : u32 maxDeviceNum;
1669 5 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1670 5 : CHK_PRT_RET(
1671 : deviceLogicId_ >= maxDeviceNum,
1672 : HCCL_ERROR(
1673 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1674 : maxDeviceNum),
1675 : HCCL_E_INTERNAL);
1676 :
1677 5 : std::string tag;
1678 5 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1679 5 : u32 index = 0;
1680 5 : ProfilerBase::GetSubmittedOpCnt(index);
1681 5 : TaskInfo tmpTaskInfo(streamID, taskID, tag, para);
1682 5 : tmpTaskInfo.index = index;
1683 5 : CHK_RET(InsertTaskMap(streamID, tmpTaskInfo));
1684 0 : CHK_RET(InsertRankInfo(tag));
1685 0 : CHK_RET(InsertOpData(tag));
1686 0 : return HCCL_SUCCESS;
1687 5 : }
1688 :
1689 2 : HcclResult TaskExceptionHandler::Save(u32 streamID, u32 taskID, const TaskParaAiv& para)
1690 : {
1691 2 : return Save(streamID, streamID, taskID, para);
1692 : }
1693 :
1694 9 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, TaskType& taskType, const TaskParaReduce& para)
1695 : {
1696 9 : return Save(streamID, streamID, taskID, taskType, para);
1697 : }
1698 :
1699 : HcclResult
1700 6 : TaskExceptionHandler::Save(u32 captureStreamID, u32 streamID, u32 taskID, const void* descBuf, size_t descBufLen)
1701 : {
1702 : u32 maxDeviceNum;
1703 6 : CHK_RET(GetMaxDevNum(maxDeviceNum));
1704 6 : CHK_PRT_RET(
1705 : deviceLogicId_ >= maxDeviceNum,
1706 : HCCL_ERROR(
1707 : "[TaskExceptionHandler][Save]deviceLogicId_[%u] is bigger than maxDeviceNum[%u]", deviceLogicId_,
1708 : maxDeviceNum),
1709 : HCCL_E_INTERNAL);
1710 6 : HCCL_INFO("[TaskExceptionHandler][%s]Save task info, streamId[%u], taskId[%u]", __func__, streamID, taskID);
1711 6 : std::string tag;
1712 6 : CHK_RET(ProfilerBase::GetTagByStream(captureStreamID, tag));
1713 6 : AlgType algType = AlgType::Reserved();
1714 6 : CHK_RET(ProfilerBase::GetAlgTypeByStream(captureStreamID, algType));
1715 6 : u32 index = 0;
1716 6 : ProfilerBase::GetSubmittedOpCnt(index);
1717 :
1718 6 : if (GetExternalInputTaskExceptionSwitch() == 1) {
1719 3 : CHK_RET(InsertOpCtxInfo(streamID, taskID, tag, algType, index, descBuf, descBufLen));
1720 : } else {
1721 3 : CHK_RET(InsertOpMap(streamID, taskID, tag, algType, index));
1722 : }
1723 6 : CHK_RET(InsertRankInfo(tag));
1724 6 : CHK_RET(InsertOpData(tag));
1725 6 : return HCCL_SUCCESS;
1726 6 : }
1727 :
1728 6 : HcclResult TaskExceptionHandler::Save(u32& streamID, u32& taskID, const void* descBuf, size_t descBufLen)
1729 : {
1730 6 : return Save(streamID, streamID, taskID, descBuf, descBufLen);
1731 : }
1732 :
1733 0 : HcclResult TaskExceptionHandler::SaveToLog(const TaskParaHost& paraHost)
1734 : {
1735 : (void)paraHost;
1736 0 : return HCCL_SUCCESS;
1737 : }
1738 :
1739 52 : HcclResult TaskExceptionHandler::InsertTaskMap(u32& streamID, TaskInfo& tmpTaskInfo) const
1740 : {
1741 52 : std::unique_lock<std::mutex> lock(taskMapMutex[deviceLogicId_]);
1742 52 : auto it = taskMap[deviceLogicId_].find(streamID);
1743 52 : if (it == taskMap[deviceLogicId_].end()) {
1744 : // streamID 复用且不会超过最大stream数量，因此Map的size超过最大stream数量属于异常场景
1745 47 : HCCL_INFO(
1746 : "streamID is [%u], deviceLogicId is [%u], taskMap size is [%u]", streamID, deviceLogicId_,
1747 : taskMap[deviceLogicId_].size());
1748 47 : CHK_PRT_RET(
1749 : taskMap[deviceLogicId_].size() >= maxStrCount,
1750 : HCCL_ERROR(
1751 : "[Insert][TaskMap]taskMap size is "
1752 : "bigger than max stream count[%u]. stream add fail",
1753 : maxStrCount),
1754 : HCCL_E_INTERNAL);
1755 1 : std::shared_ptr<deque<TaskInfo>> tmpTaskInfoQue = nullptr;
1756 1 : EXCEPTION_CATCH((tmpTaskInfoQue = make_shared<deque<TaskInfo>>()), return HCCL_E_PTR);
1757 1 : tmpTaskInfoQue->push_back(tmpTaskInfo);
1758 1 : taskMap[deviceLogicId_].insert({streamID, tmpTaskInfoQue});
1759 1 : } else { // 由于不允许多线程对同一stream操作，因此此处不需要保留锁，并且此处访问量最多，性能考虑也最好不要加锁
1760 5 : lock.unlock();
1761 5 : it->second->push_back(tmpTaskInfo);
1762 5 : if (it->second->size() > maxTaskCount) {
1763 0 : it->second->pop_front();
1764 : }
1765 : }
1766 6 : return HCCL_SUCCESS;
1767 52 : }
1768 : HcclResult
1769 3 : TaskExceptionHandler::InsertOpMap(u32& streamID, u32& taskID, string& tag, AlgType& algType, u32& index) const
1770 : {
1771 3 : FFTSOpInfo tmpOpPara;
1772 6 : char* tmpAddr = new (std::nothrow) char[tag.size() + 1]();
1773 3 : CHK_PTR_NULL(tmpAddr);
1774 3 : tmpOpPara.tag.reset(tmpAddr, default_delete<char[]>());
1775 3 : CHK_SAFETY_FUNC_RET(memcpy_sp(tmpOpPara.tag.get(), tag.size() + 1, tag.data(), tag.size()));
1776 3 : tmpOpPara.streamID = streamID;
1777 3 : tmpOpPara.taskID = taskID;
1778 3 : tmpOpPara.algType = algType;
1779 3 : tmpOpPara.index = index;
1780 3 : std::unique_lock<std::mutex> lock(opMapMutex[deviceLogicId_]); // 防止存入和读取冲突
1781 3 : auto it = opMap[deviceLogicId_].find(streamID);
1782 3 : if (it == opMap[deviceLogicId_].end()) {
1783 1 : CHK_PRT_RET(
1784 : opMap[deviceLogicId_].size() >= maxStrCount,
1785 : HCCL_ERROR(
1786 : "[Insert][OpMap]Map size is "
1787 : "bigger than max stream count[%u]. stream add fail",
1788 : maxStrCount),
1789 : HCCL_E_INTERNAL);
1790 1 : std::shared_ptr<deque<FFTSOpInfo>> tmpOpInfoQue = nullptr;
1791 1 : EXCEPTION_CATCH((tmpOpInfoQue = make_shared<deque<FFTSOpInfo>>()), return HCCL_E_PTR);
1792 1 : tmpOpInfoQue->push_back(tmpOpPara);
1793 1 : opMap[deviceLogicId_].insert({streamID, tmpOpInfoQue});
1794 1 : } else {
1795 2 : it->second->push_back(tmpOpPara);
1796 2 : if (it->second->size() > maxTaskCount) {
1797 0 : it->second->pop_front();
1798 : }
1799 : }
1800 3 : return HCCL_SUCCESS;
1801 3 : }
1802 3 : HcclResult TaskExceptionHandler::InsertOpCtxInfo(
1803 : u32& streamID, u32& taskID, string& tag, AlgType& algType, u32& index, const void* descBuf, size_t descBufLen) const
1804 : {
1805 3 : FFTSOpInfo tmpOpInfo;
1806 6 : char* tmpAddr = new (std::nothrow) char[tag.size() + 1]();
1807 3 : CHK_PTR_NULL(tmpAddr);
1808 3 : tmpOpInfo.tag.reset(tmpAddr, default_delete<char[]>());
1809 3 : CHK_SAFETY_FUNC_RET(memcpy_sp(tmpOpInfo.tag.get(), tag.size() + 1, tag.data(), tag.size()));
1810 3 : tmpOpInfo.streamID = streamID;
1811 3 : tmpOpInfo.taskID = taskID;
1812 3 : tmpOpInfo.algType = algType;
1813 3 : tmpOpInfo.index = index;
1814 3 : if (descBuf != nullptr && descBufLen > 0) {
1815 130 : char* tmpDescBuf = new (std::nothrow) char[descBufLen + 1]();
1816 1 : CHK_PTR_NULL(tmpDescBuf);
1817 1 : tmpOpInfo.descBuf.reset(tmpDescBuf, default_delete<char[]>());
1818 1 : CHK_SAFETY_FUNC_RET(memcpy_sp(tmpOpInfo.descBuf.get(), descBufLen + 1, descBuf, descBufLen));
1819 1 : tmpOpInfo.descBufLen = descBufLen;
1820 : }
1821 3 : std::shared_ptr<FFTSOpInfo> tmpOpInfoPtr = nullptr;
1822 3 : EXCEPTION_CATCH((tmpOpInfoPtr = std::make_shared<FFTSOpInfo>()), return HCCL_E_PTR);
1823 3 : *tmpOpInfoPtr = tmpOpInfo;
1824 3 : std::shared_ptr<vector<CtxInfo>> tempCtxVectorPtr = nullptr;
1825 3 : EXCEPTION_CATCH((tempCtxVectorPtr = std::make_shared<vector<CtxInfo>>()), return HCCL_E_PTR);
1826 3 : std::unique_lock<std::mutex> lock(ctxInfoVectorMutex[deviceLogicId_]); // 防止存入和读取冲突
1827 3 : *tempCtxVectorPtr = ctxInfoArray[deviceLogicId_];
1828 3 : auto tempPair = std::make_pair(tmpOpInfoPtr, tempCtxVectorPtr);
1829 3 : std::unique_lock<std::mutex> infoLock(opCtxInfoMutex[deviceLogicId_]); // 防止存入和读取冲突
1830 3 : auto tempDeque = opCtxInfo[deviceLogicId_].find(streamID);
1831 3 : if (tempDeque == opCtxInfo[deviceLogicId_].end()) {
1832 2 : CHK_PRT_RET(
1833 : opCtxInfo[deviceLogicId_].size() >= maxStrCount,
1834 : HCCL_ERROR(
1835 : "[Insert][opCtxInfo]Map size is "
1836 : "bigger than max stream count[%u]. stream add fail",
1837 : maxStrCount),
1838 : HCCL_E_INTERNAL);
1839 : std::shared_ptr<std::deque<std::pair<std::shared_ptr<FFTSOpInfo>, std::shared_ptr<std::vector<CtxInfo>>>>>
1840 2 : tmpOpInfoQue = nullptr;
1841 2 : EXCEPTION_CATCH(
1842 : (tmpOpInfoQue = std::make_shared<
1843 : std::deque<std::pair<std::shared_ptr<FFTSOpInfo>, std::shared_ptr<std::vector<CtxInfo>>>>>()),
1844 : return HCCL_E_PTR);
1845 2 : tmpOpInfoQue->push_back(tempPair);
1846 2 : opCtxInfo[deviceLogicId_].insert({streamID, tmpOpInfoQue});
1847 2 : } else {
1848 1 : tempDeque->second->push_back(tempPair);
1849 1 : if (tempDeque->second->size() > maxTaskCount) {
1850 0 : tempDeque->second->pop_front();
1851 : }
1852 : }
1853 3 : ctxInfoArray[deviceLogicId_].clear();
1854 3 : return HCCL_SUCCESS;
1855 3 : }
1856 :
1857 6 : HcclResult TaskExceptionHandler::InsertRankInfo(std::string& tag) const
1858 : {
1859 6 : std::string groupName;
1860 6 : CHK_RET(ProfilerBase::GetGroupNameByTag(tag, groupName));
1861 6 : GroupRankInfo groupRankInfo;
1862 6 : CHK_RET(ProfilerBase::GetRankInfoByGroup(groupName, groupRankInfo));
1863 6 : std::string groupUdi;
1864 6 : CHK_RET(ProfilerBase::GetUdiByGroup(groupName, groupUdi));
1865 :
1866 6 : HCCL_DEBUG("[TaskExceptionHandler][Callback]InsertRankInfo tag %s group %s", tag.c_str(), groupName.c_str());
1867 : {
1868 6 : std::unique_lock<std::mutex> groupRankMapLock(groupRankMapMutex[deviceLogicId_]);
1869 6 : std::shared_ptr<GroupRankInfo> tmpRankInfo = nullptr;
1870 6 : EXCEPTION_CATCH((tmpRankInfo = std::make_shared<GroupRankInfo>()), return HCCL_E_PTR);
1871 6 : *tmpRankInfo = groupRankInfo;
1872 6 : auto groupRankIt = groupRankMap[deviceLogicId_].find(tag);
1873 6 : if (groupRankIt == groupRankMap[deviceLogicId_].end()) {
1874 2 : auto tempPair = std::make_pair(groupName, tmpRankInfo);
1875 2 : groupRankMap[deviceLogicId_].insert({tag, tempPair});
1876 2 : } else {
1877 4 : groupRankIt->second.second = tmpRankInfo;
1878 : }
1879 6 : }
1880 :
1881 : {
1882 6 : std::lock_guard<std::mutex> groupUdiMapLock(groupUdiMapMutex[deviceLogicId_]);
1883 6 : auto groupUdiIt = groupUdiMap[deviceLogicId_].find(groupName);
1884 6 : if (groupUdiIt == groupUdiMap[deviceLogicId_].end()) {
1885 2 : groupUdiMap[deviceLogicId_].insert({groupName, groupUdi});
1886 : } else {
1887 4 : groupUdiIt->second = groupUdi;
1888 : }
1889 6 : }
1890 :
1891 6 : return HCCL_SUCCESS;
1892 6 : }
1893 :
1894 6 : HcclResult TaskExceptionHandler::InsertOpData(std::string& tag) const
1895 : {
1896 6 : OpDataInfo opDataInfo;
1897 6 : CHK_RET(ProfilerBase::GetOpDataInfoByTag(tag, opDataInfo));
1898 6 : std::unique_lock<std::mutex> lock(tagOpDataMapMutex[deviceLogicId_]);
1899 6 : auto tempDeque = tagOpDataMap[deviceLogicId_].find(tag);
1900 6 : if (tempDeque == tagOpDataMap[deviceLogicId_].end()) {
1901 2 : std::shared_ptr<queue<OpDataInfo>> tmpOpDataInfo = nullptr;
1902 2 : EXCEPTION_CATCH((tmpOpDataInfo = std::make_shared<queue<OpDataInfo>>()), return HCCL_E_PTR);
1903 2 : tmpOpDataInfo->push(opDataInfo);
1904 2 : tagOpDataMap[deviceLogicId_].insert({tag, tmpOpDataInfo});
1905 2 : HCCL_DEBUG("[TaskExceptionHandler][Callback]InsertOpData index %u tag %s", opDataInfo.index, tag.c_str());
1906 2 : } else {
1907 4 : HCCL_DEBUG(
1908 : "[TaskExceptionHandler][Callback]InsertOpData index %u opData index %u size %u tag %s", opDataInfo.index,
1909 : tempDeque->second->back().index, (tempDeque->second)->size(), tag.c_str());
1910 4 : if (tempDeque->second->back().index != opDataInfo.index) { // 需要去重，taskid不同时可能是同一个
1911 0 : tempDeque->second->push(opDataInfo);
1912 : }
1913 4 : if ((tempDeque->second)->size() > 3000) { // 队列深度大于3000则老化
1914 0 : HCCL_DEBUG("[Insert][opDataMap]Map size is [%u], need to pop head data.", (tempDeque->second)->size());
1915 0 : tempDeque->second->pop();
1916 : }
1917 : }
1918 6 : return HCCL_SUCCESS;
1919 6 : }
1920 :
1921 13 : HcclResult TaskExceptionHandler::Flush() { return HCCL_SUCCESS; }
1922 :
1923 0 : HcclResult TaskExceptionHandler::TaskExceptionHandler::Run(const StepData& stepData)
1924 : {
1925 : (void)stepData;
1926 0 : return HCCL_SUCCESS;
1927 : }
|
