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