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 "opretry_manager.h"
12 : #include "opretry_link_manage.h"
13 : #include "opretry_connection_pub.h"
14 : #include "opretry_agent.h"
15 : #include "opretry_server.h"
16 : #include "externalinput_pub.h"
17 : #include "adapter_rts_common.h"
18 : #include "sal_pub.h"
19 :
20 : namespace hccl {
21 0 : HcclResult OpRetryManager::Init()
22 : {
23 0 : CHK_PRT_RET(initialized_ == true, HCCL_WARNING("OpRetryManager has already initialized"), HCCL_SUCCESS);
24 0 : initialized_ = true;
25 0 : HCCL_INFO("OpRetryManager Init success");
26 0 : return HCCL_SUCCESS;
27 : }
28 :
29 0 : HcclResult OpRetryManager::RegisterOpRetryMachine(OpRetryAgentParam &agentParam, u32 rankSize, bool isRoot,
30 : std::map<u32, std::shared_ptr<HcclSocket> > &serverConnections, const OpRetryServerInfo& serverInfo)
31 : {
32 0 : std::unique_lock<std::mutex> lock(ProcessLock_);
33 0 : CHK_SMART_PTR_NULL(agentParam.h2dPtr);
34 0 : CHK_SMART_PTR_NULL(agentParam.d2hPtr);
35 0 : CHK_SMART_PTR_NULL(agentParam.opStreamPtr);
36 0 : CHK_PRT_RET(agentParam.group.empty(),
37 : HCCL_ERROR("[OpRetryManager][RegisterOpRetryMachine]params invalid, group is empty"), HCCL_E_PARA);
38 0 : if (agentParam.agentConnection == nullptr && serverConnections.empty()) {
39 0 : CHK_RET(OpRetryConnectionPub::Init(agentParam.group, rankSize, serverInfo, agentParam.agentInfo));
40 0 : CHK_RET(OpRetryConnectionPub::GetConns(agentParam.group, isRoot, agentParam.agentConnection, serverConnections));
41 : }
42 : // 初始化
43 0 : if (initialized_ == false) {
44 0 : CHK_RET(Init());
45 : }
46 :
47 : // 注册agent状态机
48 0 : CHK_RET(RegisterAgentRetryMachine(agentParam));
49 :
50 : // 注册server状态机
51 0 : if (isRoot) {
52 0 : CHK_RET(RegisterServerRetryMachine(agentParam.group, serverConnections, agentParam.agentInfo));
53 : }
54 0 : HCCL_INFO("[Register][RetryMachine]group[%s] register success", agentParam.group.c_str());
55 0 : return HCCL_SUCCESS;
56 0 : }
57 :
58 0 : HcclResult OpRetryManager::RegisterAgentRetryMachine(OpRetryAgentParam &agentParam)
59 : {
60 0 : std::string &group = agentParam.group;
61 0 : if (agentOpRetry_.find(group) != agentOpRetry_.end()) {
62 0 : HCCL_INFO("[Register][AgentRetryMachine]group[%s] has Registered to agentOpRetry, skip", group.c_str());
63 0 : return HCCL_SUCCESS;
64 : }
65 :
66 0 : RetryCtrl retryCtrl;
67 0 : agentOpRetry_.insert(std::make_pair(group, std::move(retryCtrl)));
68 0 : std::shared_ptr<OpRetryBase> retryPtr;
69 0 : EXCEPTION_CATCH((retryPtr = std::make_shared<OpRetryAgentRunning>()), return HCCL_E_PTR);
70 0 : EXCEPTION_CATCH((agentOpRetry_[group].retryCtx =
71 : std::make_shared<RetryContext>(agentParam, retryPtr)), return HCCL_E_PTR);
72 0 : agentOpRetry_[group].startExec = true;
73 0 : agentOpRetry_[group].retryCtx->SetRetryState(RETRY_STATE_AGENT_RUNNING, retryPtr);
74 :
75 0 : aclrtContext ctx = nullptr;
76 0 : CHK_RET(hrtCtxGetCurrent(&ctx));
77 0 : agentOpRetry_[group].thread.reset(new (std::nothrow) std::thread(&OpRetryManager::RetryStateMonitor, this,
78 0 : group, agentOpRetry_[group].retryCtx, std::ref(agentOpRetry_[group].startExec), ctx));
79 0 : CHK_SMART_PTR_NULL(agentOpRetry_[group].thread);
80 0 : HCCL_INFO("[%s]group[%s] rank[%u], register to agentOpRetry success",
81 : __func__, group.c_str(), agentParam.agentInfo.userRank);
82 0 : return HCCL_SUCCESS;
83 0 : }
84 :
85 0 : HcclResult OpRetryManager::RegisterServerRetryMachine(const std::string& group,
86 : std::map<u32, std::shared_ptr<HcclSocket>> &serverConnections, const OpRetryAgentInfo& agentInfo)
87 : {
88 0 : if (serverOpRetry.find(group) != serverOpRetry.end()) {
89 0 : HCCL_INFO("[Register][ServerRetryMachine]group[%s] has Registered to serverOpRetry, skip", group.c_str());
90 0 : return HCCL_SUCCESS;
91 : }
92 0 : for (auto it = serverConnections.begin(); it != serverConnections.end(); ++it) {
93 0 : CHK_SMART_PTR_NULL(it->second);
94 : }
95 :
96 0 : RetryCtrl retryCtrl;
97 0 : serverOpRetry.insert(std::make_pair(group, std::move(retryCtrl)));
98 0 : std::shared_ptr<OpRetryBase> retryPtr = nullptr;
99 0 : EXCEPTION_CATCH((retryPtr = std::make_shared<OpRetryServerRunning>()), return HCCL_E_PTR);
100 :
101 0 : EXCEPTION_CATCH((serverOpRetry[group].retryCtx =
102 : std::make_shared<RetryContext>(serverConnections, retryPtr, agentInfo)), return HCCL_E_PTR);
103 0 : serverOpRetry[group].startExec = true;
104 0 : serverOpRetry[group].retryCtx->SetRetryState(RETRY_STATE_SERVER_RUNNING, retryPtr);
105 :
106 0 : HcclRtContext ctx = nullptr;
107 0 : CHK_RET(hrtCtxGetCurrent(&ctx));
108 0 : serverOpRetry[group].thread.reset(new (std::nothrow) std::thread(&OpRetryManager::RetryStateMonitor, this,
109 0 : group, serverOpRetry[group].retryCtx, std::ref(serverOpRetry[group].startExec), ctx));
110 0 : CHK_SMART_PTR_NULL(serverOpRetry[group].thread);
111 0 : HCCL_INFO("[%s]group[%s] rank[%u], register to serverOpRetry success", __func__, group.c_str(), agentInfo.userRank);
112 0 : return HCCL_SUCCESS;
113 0 : }
114 :
115 0 : HcclResult OpRetryManager::UnRegisterOpRetryManager(const std::string& group)
116 : {
117 0 : std::unique_lock<std::mutex> lock(ProcessLock_);
118 0 : CHK_PRT_RET(group.empty(),
119 : HCCL_ERROR("[OpRetryManager][UnRegisterOpRetryManager]params invalid, group is empty"), HCCL_E_PARA);
120 0 : HCCL_INFO("[UnRegister][OpRetryManager]group[%s] unregister start", group.c_str());
121 0 : CHK_PRT_RET(initialized_ == false, HCCL_WARNING("OpRetryManager has been destroyed"), HCCL_SUCCESS);
122 :
123 0 : if (agentOpRetry_.find(group) != agentOpRetry_.end()) {
124 0 : agentOpRetry_[group].startExec = false;
125 0 : if (agentOpRetry_[group].thread != nullptr && agentOpRetry_[group].thread->joinable()) {
126 0 : agentOpRetry_[group].thread->join();
127 : }
128 0 : agentOpRetry_.erase(group);
129 0 : HCCL_INFO("[UnRegister][OpRetryManager]group[%s] unregister agentOpRetry success", group.c_str());
130 : }
131 :
132 0 : if (serverOpRetry.find(group) != serverOpRetry.end()) {
133 0 : serverOpRetry[group].startExec = false;
134 0 : if (serverOpRetry[group].thread != nullptr && serverOpRetry[group].thread->joinable()) {
135 0 : serverOpRetry[group].thread->join();
136 : }
137 0 : serverOpRetry.erase(group);
138 0 : HCCL_INFO("[UnRegister][OpRetryManager]group[%s] unregister serverOpRetry success", group.c_str());
139 : }
140 0 : HCCL_INFO("[UnRegister][OpRetryManager]group[%s] unregister success", group.c_str());
141 0 : OpRetryConnectionPub::DeInit(group);
142 0 : return HCCL_SUCCESS;
143 0 : }
144 :
145 0 : void OpRetryManager::RetryStateMonitor(const std::string &group, std::shared_ptr<RetryContext> retryCtx,
146 : const bool &startExec, HcclRtContext rtCtx)
147 : {
148 0 : CHK_SMART_PTR_RET_NULL(retryCtx);
149 0 : CHK_SMART_PTR_RET_NULL(rtCtx);
150 0 : CHK_RET_NULL(hrtCtxSetCurrent(rtCtx));
151 :
152 : // 给当前线程添加名字
153 0 : SetThreadName("Hccl_OpRetry");
154 :
155 0 : HCCL_RUN_INFO("[%s]%s start, group[%s], rankId[%u], IpInfo[%s]", __func__, retryCtx->GetOpRetryMachineType(),
156 : group.c_str(), retryCtx->GetRankId(), retryCtx->GetDfxIpInfo());
157 :
158 0 : HcclResult ret = HCCL_SUCCESS;
159 0 : while(initialized_ && startExec) {
160 0 : ret = retryCtx->Request();
161 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
162 : HCCL_ERROR("RetryStateMonitor group[%s] exec fail", group.c_str()), );
163 : }
164 0 : HCCL_INFO("RetryStateMonitor group[%s] exit, ret[%d], initialized_[%d], startExec[%d]",
165 : group.c_str(), ret, initialized_, startExec);
166 : }
167 :
168 0 : HcclResult OpRetryManager::AddLinkInfoByIdentifier(s32 deviceLogicID, const std::string &identifier,
169 : const std::string &newTag, std::vector<u32> &remoteRankList, bool incre)
170 : {
171 0 : return OpretryLinkManage::GetInstance(deviceLogicID).AddLinkInfoByIdentifier(identifier, newTag, remoteRankList, incre);
172 : }
173 :
174 0 : HcclResult OpRetryManager::GetLinkInfoByIdentifier(s32 deviceLogicID, const std::string &identifier,
175 : const std::string &newTag, std::vector<u32> &remoteRankList, bool isGetGroupAllRemoteRank)
176 : {
177 0 : if (isGetGroupAllRemoteRank) {
178 0 : return OpretryLinkManage::GetInstance(deviceLogicID).GetLinkInfoByIdentifier(identifier, remoteRankList);
179 : }
180 0 : return OpretryLinkManage::GetInstance(deviceLogicID).GetLinkInfoByIdentifier(identifier, newTag, remoteRankList);
181 : }
182 :
183 0 : HcclResult OpRetryManager::DeleteLinkInfoByIdentifier(s32 deviceLogicID, const std::string &identifier)
184 : {
185 0 : return OpretryLinkManage::GetInstance(deviceLogicID).DeleteLinkInfoByIdentifier(identifier);
186 : }
187 :
188 0 : HcclResult OpRetryManager::SetRetryStateToWaitResume(const std::string &group, bool isRoot)
189 : {
190 0 : std::unique_lock<std::mutex> lock(ProcessLock_);
191 0 : std::chrono::seconds setTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
192 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
193 0 : if (agentOpRetry_.find(group) != agentOpRetry_.end()) {
194 0 : agentOpRetry_[group].retryCtx->isAgentStateWaitResume_ = true;
195 0 : while (agentOpRetry_[group].retryCtx->GetRetryState() != RETRY_STATE_AGENT_WAIT_RESUME) {
196 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
197 0 : const auto setTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
198 0 : if (setTime > setTimeout) {
199 0 : HCCL_ERROR("[OpRetryManager][SetRetryStateToWaitResume]group[%s], set agent state to wait resume timeout.", group.c_str());
200 0 : return HCCL_E_TIMEOUT;
201 : }
202 0 : if (agentOpRetry_[group].retryCtx->isOpRetryQuit) {
203 0 : agentOpRetry_[group].retryCtx->isOpRetryQuit = false;
204 0 : break;
205 : }
206 : }
207 : }
208 :
209 0 : if (isRoot && serverOpRetry.find(group) != serverOpRetry.end()) {
210 0 : serverOpRetry[group].retryCtx->isServerStateWaitResume_ = true;
211 0 : while (serverOpRetry[group].retryCtx->GetRetryState() != RETRY_STATE_SERVER_WAIT_RESUME) {
212 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
213 0 : const auto setTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
214 0 : if (setTime > setTimeout) {
215 0 : HCCL_ERROR("[OpRetryManager][SetRetryStateToWaitResume]group[%s], set server state to wait resume timeout.", group.c_str());
216 0 : return HCCL_E_TIMEOUT;
217 : }
218 0 : if (serverOpRetry[group].retryCtx->isOpRetryQuit) {
219 0 : serverOpRetry[group].retryCtx->isOpRetryQuit = false;
220 0 : break;
221 : }
222 : }
223 : }
224 0 : HCCL_INFO("[OpRetryManager][SetRetryStateToWaitResume]group[%s], set state to wait resume success", group.c_str());
225 0 : return HCCL_SUCCESS;
226 0 : }
227 :
228 0 : HcclResult OpRetryManager::ExitWaitResumeState(const std::string &group, bool isRoot, bool haveCommEnableBackupLink, bool &isChangedLink)
229 : {
230 0 : HCCL_RUN_INFO("[OpRetryManager][ExitWaitResumeState]group[%s], haveCommEnableBackupLink[%d] exit wait resume state start", group.c_str(), haveCommEnableBackupLink);
231 0 : std::unique_lock<std::mutex> lock(ProcessLock_);
232 0 : std::chrono::seconds exitTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
233 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
234 0 : if (isRoot && serverOpRetry.find(group) != serverOpRetry.end()) {
235 0 : serverOpRetry[group].retryCtx->haveCommEnableBackupLink_ = haveCommEnableBackupLink;
236 : std::atomic_thread_fence(std::memory_order_seq_cst);
237 0 : serverOpRetry[group].retryCtx->isServerStateWaitResume_ = false;
238 : }
239 0 : if (agentOpRetry_.find(group) != agentOpRetry_.end()) {
240 0 : agentOpRetry_[group].retryCtx->haveCommEnableBackupLink_ = haveCommEnableBackupLink;
241 : std::atomic_thread_fence(std::memory_order_seq_cst);
242 0 : agentOpRetry_[group].retryCtx->isAgentStateWaitResume_ = false;
243 : }
244 0 : while (isRoot && serverOpRetry.find(group) != serverOpRetry.end() && serverOpRetry[group].retryCtx->GetRetryState() != RETRY_STATE_SERVER_RUNNING) {
245 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
246 0 : const auto exitTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
247 0 : if (exitTime > exitTimeout) {
248 0 : HCCL_ERROR("[OpRetryManager][ExitWaitResumeState]group[%s], state[%d], server exit wait resume state timeout", group.c_str(), serverOpRetry[group].retryCtx->GetRetryState());
249 0 : return HCCL_E_TIMEOUT;
250 : }
251 0 : if (haveCommEnableBackupLink) {
252 0 : isChangedLink = true;
253 : }
254 : }
255 0 : while (agentOpRetry_.find(group) != agentOpRetry_.end() && agentOpRetry_[group].retryCtx->GetRetryState() != RETRY_STATE_AGENT_RUNNING) {
256 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
257 0 : const auto exitTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
258 0 : if (exitTime > exitTimeout) {
259 0 : HCCL_ERROR("[OpRetryManager][ExitWaitResumeState]group[%s], state[%d], agent exit wait resume state timeout", group.c_str(), agentOpRetry_[group].retryCtx->GetRetryState());
260 0 : return HCCL_E_TIMEOUT;
261 : }
262 0 : if (agentOpRetry_[group].retryCtx->isRecivedCmdToCheckLink) {
263 0 : isChangedLink = true;
264 : }
265 : }
266 0 : HCCL_RUN_INFO("[OpRetryManager][ExitWaitResumeState]group[%s], exit wait resume state success, isChangedLink[%d]", group.c_str(), isChangedLink);
267 0 : return HCCL_SUCCESS;
268 0 : }
269 :
270 0 : bool OpRetryManager::IsPaused(const std::string &group)
271 : {
272 0 : return (serverOpRetry.find(group) == serverOpRetry.end()
273 0 : || serverOpRetry[group].retryCtx == nullptr
274 0 : || serverOpRetry[group].retryCtx->IsPaused())
275 0 : && (agentOpRetry_.find(group) == agentOpRetry_.end()
276 0 : || agentOpRetry_[group].retryCtx == nullptr
277 0 : || agentOpRetry_[group].retryCtx->IsPaused());
278 : }
279 :
280 0 : bool OpRetryManager::IsResumed(const std::string &group)
281 : {
282 0 : return (serverOpRetry.find(group) == serverOpRetry.end()
283 0 : || serverOpRetry[group].retryCtx == nullptr
284 0 : || !serverOpRetry[group].retryCtx->IsPaused())
285 0 : && (agentOpRetry_.find(group) == agentOpRetry_.end()
286 0 : || agentOpRetry_[group].retryCtx == nullptr
287 0 : || !agentOpRetry_[group].retryCtx->IsPaused());
288 : }
289 : }
|
