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 : #include <algorithm>
11 : #include <mutex>
12 : #include "task_abort_handler.h"
13 : #include "ccu_dev_mgr.h"
14 : #include "log.h"
15 :
16 : namespace Hccl {
17 : using HcclUs = std::chrono::steady_clock::time_point;
18 : static std::mutex vecMutex;
19 4 : static int32_t TaskAbortPre(const std::vector<HcclCommunicator *> &commVector,
20 : const std::chrono::seconds &localtimeout)
21 : {
22 4 : HcclResult ret = HCCL_SUCCESS;
23 4 : bool isUseTimeOut = localtimeout != std::chrono::seconds(0);
24 4 : std::chrono::seconds elapsed{};
25 8 : for (const auto& comm : commVector) {
26 6 : if (isUseTimeOut) {
27 : std::chrono::steady_clock::time_point startTime =
28 3 : std::chrono::steady_clock::now();
29 3 : ret = comm->Suspend();
30 3 : elapsed = std::chrono::duration_cast<std::chrono::seconds>(
31 6 : std::chrono::steady_clock::now() - startTime);
32 : } else {
33 3 : ret = comm->Suspend();
34 : }
35 6 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
36 6 : HCCL_ERROR("[NsRecovery] finish suspend failed");
37 2 : return static_cast<int>(TaskAbortResult::TASK_ABORT_FAIL);
38 : }
39 12 : HCCL_DEBUG("[NsRecovery]finish suspend success");
40 4 : if (isUseTimeOut) {
41 2 : CHK_PRT_RET(
42 : elapsed > localtimeout,
43 : HCCL_ERROR("[NsRecovery][suspend] NsRecovery suspend timeOut"),
44 : static_cast<int>(TaskAbortResult::TASK_ABORT_TIMEOUT));
45 : }
46 : }
47 2 : return static_cast<int>(TaskAbortResult::TASK_ABORT_SUCCESS);
48 : }
49 :
50 4 : static int32_t TaskAbortPost(const std::vector<HcclCommunicator *> &commVector,
51 : int32_t deviceLogicId,
52 : const std::chrono::seconds &localtimeout) {
53 4 : HcclResult ret = HCCL_SUCCESS;
54 4 : bool isUseTimeOut = localtimeout != std::chrono::seconds(0);
55 4 : std::chrono::seconds elapsed{};
56 4 : if (CcuIsInited(deviceLogicId)) {
57 0 : CHK_RET(HcclCcuTaskKillPreProcess(deviceLogicId));
58 : } else {
59 12 : HCCL_INFO("[NsRecovery][Callback] CCU not inited, skip TaskKillPreProcess, deviceLogicId[%d]", deviceLogicId);
60 : }
61 8 : for (const auto& comm : commVector) {
62 6 : if (isUseTimeOut) {
63 : std::chrono::steady_clock::time_point startTime =
64 3 : std::chrono::steady_clock::now();
65 3 : ret = comm->Clean();
66 3 : elapsed = std::chrono::duration_cast<std::chrono::seconds>(
67 6 : std::chrono::steady_clock::now() - startTime);
68 : } else {
69 3 : ret = comm->Clean();
70 : }
71 6 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
72 6 : HCCL_ERROR("[NsRecovery][Callback] finish clean failed");
73 2 : return static_cast<int>(TaskAbortResult::TASK_ABORT_FAIL);
74 : }
75 12 : HCCL_INFO("[NsRecovery][Callback] finish clean success");
76 4 : if (isUseTimeOut) {
77 2 : CHK_PRT_RET(elapsed > localtimeout,
78 : HCCL_ERROR("[NsRecovery][Callback] NsRecovery Clean timeout"),
79 : static_cast<int>(TaskAbortResult::TASK_ABORT_TIMEOUT));
80 : }
81 : }
82 2 : if (CcuIsInited(deviceLogicId)) {
83 0 : CHK_RET(HcclCcuTaskKillPostProcess(deviceLogicId));
84 : } else {
85 6 : HCCL_INFO("[NsRecovery][Callback] CCU not inited, skip TaskKillPostProcess, deviceLogicId[%d]", deviceLogicId);
86 : }
87 2 : return static_cast<int>(TaskAbortResult::TASK_ABORT_SUCCESS);
88 : }
89 :
90 8 : int32_t ProcessTaskAbortHandleCallback(int32_t deviceLogicId, aclrtDeviceTaskAbortStage stage, uint32_t timeout,
91 : void* args)
92 : {
93 8 : HcclUs startut = std::chrono::steady_clock::now();
94 8 : CHK_PTR_NULL(args);
95 8 : auto &commVector = *(static_cast<std::vector<HcclCommunicator *> *>(args));
96 24 : HCCL_INFO("[NsRecovery][Callback] ProcessTaskAbortHandleCallback begin, deviceLogicId [%d], stage [%d], commVector "
97 : "size [%lu]",
98 : deviceLogicId, stage, commVector.size());
99 8 : const std::chrono::seconds localtimeout = std::chrono::seconds(timeout);
100 :
101 8 : if (stage == aclrtDeviceTaskAbortStage::ACL_RT_DEVICE_TASK_ABORT_PRE) {
102 4 : auto result = TaskAbortPre(commVector, localtimeout);
103 4 : if (result != static_cast<int>(TaskAbortResult::TASK_ABORT_SUCCESS)) {
104 2 : return result;
105 : }
106 : }
107 4 : else if (stage == aclrtDeviceTaskAbortStage::ACL_RT_DEVICE_TASK_ABORT_POST) {
108 4 : auto result = TaskAbortPost(commVector, deviceLogicId, localtimeout);
109 4 : if (result != static_cast<int>(TaskAbortResult::TASK_ABORT_SUCCESS)) {
110 2 : return result;
111 : }
112 : }
113 4 : HcclUs endut = std::chrono::steady_clock::now();
114 12 : HCCL_INFO("[NsRecovery][Callback] ProcessTaskAbortHandleCallback success, take time:[%lld]us",
115 : std::chrono::duration_cast<std::chrono::microseconds>(endut - startut).count());
116 4 : return static_cast<int>(TaskAbortResult::TASK_ABORT_SUCCESS);
117 : }
118 :
119 1 : TaskAbortHandler::TaskAbortHandler()
120 : {
121 1 : std::string name = "HCCL";
122 1 : HrtDeviceAbortRegCallBack(ProcessTaskAbortHandleCallback, static_cast<void *>(&commVector), name);
123 1 : }
124 :
125 1 : TaskAbortHandler::~TaskAbortHandler()
126 : {
127 1 : std::string name = "HCCL";
128 1 : DECTOR_TRY_CATCH("TaskAbortHandler", HrtDeviceAbortRegCallBack(nullptr, nullptr, name));
129 1 : }
130 :
131 713 : TaskAbortHandler &TaskAbortHandler::GetInstance()
132 : {
133 713 : static TaskAbortHandler handler;
134 713 : return handler;
135 : }
136 :
137 273 : HcclResult TaskAbortHandler::Register(HcclCommunicator *communicator)
138 : {
139 273 : std::lock_guard<std::mutex> lock(vecMutex);
140 273 : commVector.push_back(communicator);
141 819 : HCCL_INFO("TaskAbortHandler::Register success, commVector size is [%lu]", commVector.size());
142 :
143 273 : return HCCL_SUCCESS;
144 273 : }
145 :
146 273 : HcclResult TaskAbortHandler::UnRegister(HcclCommunicator *communicator)
147 : {
148 273 : std::lock_guard<std::mutex> lock(vecMutex);
149 819 : HCCL_INFO("TaskAbortHandler::UnRegister Begin, commVector size is [%lu]", commVector.size());
150 273 : auto it = std::find(commVector.begin(), commVector.end(), communicator);
151 273 : if (it != commVector.end()) {
152 270 : commVector.erase(it);
153 : } else {
154 9 : HCCL_WARNING("TaskAbortHandler::UnRegister, comm not found.");
155 : }
156 819 : HCCL_INFO("TaskAbortHandler::UnRegister finish, commVector size is [%lu]", commVector.size());
157 273 : return HCCL_SUCCESS;
158 273 : }
159 : }
|
