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 "task_abort_handler_v1.h"
12 : #include "sal_pub.h"
13 : #include "hccl_communicator.h"
14 : #include "adapter_rts_common.h"
15 :
16 : using namespace hccl;
17 : using namespace std;
18 :
19 : static std::vector<HcclCommunicator*> commVector;
20 : static Referenced ref_;
21 : static std::mutex mutex_;
22 : struct TaskAbortCbArgs {
23 : u64 commVectorAddr;
24 : };
25 :
26 6 : TaskAbortHandler::TaskAbortHandler() {}
27 6 : TaskAbortHandler::~TaskAbortHandler() {}
28 :
29 : int32_t
30 8 : ProcessTaskAbortHandleCallback(int32_t deviceLogicId, aclrtDeviceTaskAbortStage stage, uint32_t timeout, void* args)
31 : {
32 8 : HcclUs startut = TIME_NOW();
33 8 : CHK_PTR_NULL(args);
34 8 : std::lock_guard<std::mutex> lock(mutex_);
35 8 : HCCL_INFO(
36 : "ProcessTaskAbortHandleCallback begin, deviceLogicId [%d], stage [%d], args [%p], commVector v1 size [%u], "
37 : "ref_ count is [%d]",
38 : deviceLogicId, stage, args, commVector.size(), ref_.Count());
39 8 : const std::chrono::seconds localtimeout = std::chrono::seconds(timeout);
40 8 : HcclResult ret = HCCL_SUCCESS;
41 8 : if (localtimeout != std::chrono::seconds(0)) {
42 4 : if (stage == ACL_RT_DEVICE_TASK_ABORT_PRE) {
43 0 : for (size_t i = 0; i < commVector.size(); i++) {
44 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
45 0 : ret = commVector[i]->Suspend();
46 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
47 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
48 0 : HCCL_ERROR("[NsRecovery] finish suspend failed");
49 0 : return static_cast<int>(TaskAbortResult::TaskAbort_Fail);
50 : }
51 0 : HCCL_DEBUG("[NsRecovery]finish suspend success");
52 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
53 0 : CHK_PRT_RET(
54 : elapsed > localtimeout, HCCL_ERROR("[NsRecovery][suspend] NsRecovery suspend timeOut"),
55 : static_cast<int>(TaskAbortResult::TaskAbort_TimeOut));
56 : }
57 4 : } else if (stage == ACL_RT_DEVICE_TASK_ABORT_POST) {
58 8 : for (size_t i = 0; i < commVector.size(); i++) {
59 4 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
60 4 : ret = commVector[i]->StopExec();
61 4 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
62 4 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
63 0 : HCCL_ERROR("[NsRecovery] finish stopExec failed");
64 0 : return static_cast<int>(TaskAbortResult::TaskAbort_Fail);
65 : }
66 4 : HCCL_DEBUG("[NsRecovery]finish stopExec success");
67 4 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
68 4 : CHK_PRT_RET(
69 : elapsed > localtimeout, HCCL_ERROR("[NsRecovery][StopExec] NsRecovery StopExec timeOut"),
70 : static_cast<int>(TaskAbortResult::TaskAbort_TimeOut));
71 : }
72 4 : HcclUs stopExecUt = TIME_NOW();
73 4 : HCCL_RUN_INFO(
74 : "TaskAbortHandler:ProcessTaskAbortHandleCallback, stopExec take time:[%lld]us",
75 : DURATION_US(stopExecUt - startut).count());
76 :
77 7 : for (size_t i = 0; i < commVector.size(); i++) {
78 4 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
79 4 : ret = commVector[i]->Clean();
80 4 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
81 4 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
82 1 : HCCL_ERROR("[NsRecovery] finish clean failed");
83 1 : return static_cast<int>(TaskAbortResult::TaskAbort_Fail);
84 : }
85 3 : HCCL_DEBUG("[NsRecovery]finish clean success");
86 3 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
87 3 : CHK_PRT_RET(
88 : elapsed > localtimeout, HCCL_ERROR("[NsRecovery][clean] NsRecovery Clean timeOut"),
89 : static_cast<int>(TaskAbortResult::TaskAbort_TimeOut));
90 3 : CHK_RET(commVector[i]->Stop());
91 : }
92 : }
93 : } else {
94 4 : if (stage == ACL_RT_DEVICE_TASK_ABORT_PRE) {
95 0 : for (size_t i = 0; i < commVector.size(); i++) {
96 0 : ret = commVector[i]->Suspend();
97 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
98 0 : HCCL_ERROR("[NsRecovery] finish suspend failed");
99 0 : return static_cast<int>(TaskAbortResult::TaskAbort_Fail);
100 : }
101 0 : HCCL_DEBUG("[NsRecovery]finish suspend success");
102 : }
103 4 : } else if (stage == ACL_RT_DEVICE_TASK_ABORT_POST) {
104 8 : for (size_t i = 0; i < commVector.size(); i++) {
105 4 : ret = commVector[i]->StopExec();
106 4 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
107 0 : HCCL_ERROR("[NsRecovery] finish stopExec failed");
108 1 : return static_cast<int>(TaskAbortResult::TaskAbort_Fail);
109 : }
110 4 : HCCL_DEBUG("[NsRecovery]finish stopExec success");
111 : }
112 4 : HcclUs stopExecUt = TIME_NOW();
113 4 : HCCL_RUN_INFO(
114 : "TaskAbortHandler:ProcessTaskAbortHandleCallback, stopExec take time:[%lld]us",
115 : DURATION_US(stopExecUt - startut).count());
116 7 : for (size_t i = 0; i < commVector.size(); i++) {
117 4 : ret = commVector[i]->Clean();
118 4 : if (ret != HCCL_SUCCESS && ret != HCCL_E_SUSPENDING) {
119 1 : HCCL_ERROR("[NsRecovery] finish clean failed");
120 1 : return static_cast<int>(TaskAbortResult::TaskAbort_Fail);
121 : }
122 3 : HCCL_DEBUG("[NsRecovery]finish clean success");
123 3 : CHK_RET(commVector[i]->Stop());
124 : }
125 : }
126 : }
127 :
128 6 : HcclUs endut = TIME_NOW();
129 6 : HCCL_RUN_INFO(
130 : "TaskAbortHandler:ProcessTaskAbortHandleCallback, deviceLogicId [%d], stage [%d], total take time:[%lld]us",
131 : deviceLogicId, stage, DURATION_US(endut - startut).count());
132 6 : return static_cast<int>(TaskAbortResult::TaskAbort_Success);
133 8 : }
134 :
135 417 : HcclResult TaskAbortHandler::Init(HcclCommunicator* communicator)
136 : {
137 417 : std::unique_lock<std::mutex> lock(mutex_);
138 417 : HCCL_INFO("TaskAbortHandler::Init commVector size is [%d], ref_ count is [%d]", commVector.size(), ref_.Count());
139 417 : if (ref_.Count() == 0) {
140 251 : CHK_RET(hrtTaskAbortHandleCallback(ProcessTaskAbortHandleCallback, static_cast<void*>(&commVector)));
141 : }
142 417 : ref_.Ref();
143 417 : commVector.push_back(communicator);
144 :
145 417 : return HCCL_SUCCESS;
146 417 : }
147 :
148 695 : HcclResult TaskAbortHandler::DeInit(HcclCommunicator* communicator)
149 : {
150 695 : std::unique_lock<std::mutex> lock(mutex_);
151 695 : HCCL_INFO("TaskAbortHandler::DeInit commVector size is [%d], ref_ count is [%d]", commVector.size(), ref_.Count());
152 1471 : for (auto it = commVector.begin(); it != commVector.end();) {
153 776 : if (*it == communicator) {
154 413 : it = commVector.erase(it);
155 : } else {
156 363 : ++it;
157 : }
158 : }
159 695 : ref_.Unref();
160 695 : if (ref_.Count() == 0) {
161 250 : commVector.clear();
162 250 : CHK_RET(hrtTaskAbortHandleCallback(nullptr, nullptr));
163 : }
164 695 : return HCCL_SUCCESS;
165 695 : }
|
