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