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 <chrono>
12 : #include "externalinput_pub.h"
13 : #include "sal_pub.h"
14 : #include "heartbeat.h"
15 : #include "comm_configer.h"
16 : #include "opretry_agent.h"
17 :
18 : namespace hccl {
19 5 : HcclResult CreateOpRetryAgentByState(RetryState state, RetryContext* retryCtx)
20 : {
21 5 : HCCL_INFO("[OpRetry][Agent]CreateOpRetryAgentByState state[%s]", GetReadableState(state));
22 5 : std::shared_ptr<OpRetryBase> retryPtr = nullptr;
23 5 : switch (state) {
24 2 : case RETRY_STATE_AGENT_RUNNING: {
25 2 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentRunning>(), return HCCL_E_PTR);
26 2 : break;
27 : }
28 0 : case RETRY_STATE_AGENT_RETRY_FAIL: {
29 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentRetryFail>(), return HCCL_E_PTR);
30 0 : break;
31 : }
32 1 : case RETRY_STATE_RESP_AICPU_ERR:
33 : case RETRY_STATE_RESP_AICPU_STOPED:
34 : case RETRY_STATE_RESP_STREAM_STOPED:
35 : case RETRY_STATE_RESP_STREAM_CLEARED:
36 : case RETRY_STATE_RESP_LINK_CHANGED:
37 : case RETRY_STATE_RESP_STOP_TRANSPORT:
38 : case RETRY_STATE_RESP_NOTIFY_RESETED:
39 : case RETRY_STATE_RESP_RESUME_TRANSPORT:
40 : case RETRY_STATE_RESP_CHECK_INFO:
41 : case RETRY_STATE_RESP_AICPU_RETRYEND:
42 : case RETRY_STATE_RESP_RUNNING_ERR: {
43 1 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentResponse>(), return HCCL_E_PTR);
44 1 : break;
45 : }
46 0 : case RETRY_STATE_RESP_LINK_CHECKED: {
47 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentResponseLinkInfo>(), return HCCL_E_PTR);
48 0 : break;
49 : }
50 0 : case RETRY_STATE_WAIT_CHANGE_LINK_INFO: {
51 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentWaitChangeLinkInfo>(), return HCCL_E_PTR);
52 0 : break;
53 : }
54 0 : case RETRY_STATE_WAIT_CMD_STOP_AICPU:
55 : case RETRY_STATE_WAIT_CMD_STOP_STREAM:
56 : case RETRY_STATE_WAIT_CMD_CLEAR_STREAM:
57 : case RETRY_STATE_WAIT_CMD_STOP_TRANSPORT:
58 : case RETRY_STATE_WAIT_CMD_RESET_NOTIFY:
59 : case RETRY_STATE_WAIT_CMD_CHECK_LINK:
60 : case RETRY_STATE_WAIT_CMD_RESUME_TRANSPORT:
61 : case RETRY_STATE_WAIT_CMD_CHECK:
62 : case RETRY_STATE_WAIT_CMD_CAN_RETRY:
63 : case RETRY_STATE_WAIT_CMD_RETRY_FAIL: {
64 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentWaitCmd>(), return HCCL_E_PTR);
65 0 : break;
66 : }
67 0 : case RETRY_STATE_POLL_AICPU_STOPED:
68 : case RETRY_STATE_POLL_AICPU_CHANGED:
69 : case RETRY_STATE_POLL_AICPU_RETRYEND:
70 : case RETRY_STATE_POLL_STREAM_STOPED: {
71 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentPollAicpuStop>(), return HCCL_E_PTR);
72 0 : break;
73 : }
74 : // 发送主动借轨信息
75 1 : case RETRY_STATE_SEND_SWITCH_INFO:
76 1 : EXCEPTION_CATCH(retryPtr = std::make_shared<SwitchNicAgentSendSwitchInfo>(), return HCCL_E_PTR);
77 1 : break;
78 : // 等待server RetryCommand命令
79 1 : case RETRY_STATE_WAIT_CMD_SEND_AICPU:
80 1 : EXCEPTION_CATCH(retryPtr = std::make_shared<SwitchNicAgentWaitCmd>(), return HCCL_E_PTR);
81 1 : break;
82 : // Resume 过程中检查网口和链路状态(接收来自Server的命令，检查网口和链路状态，将信息回复发送给Server)
83 0 : case RETRY_RESUME_STATE_AGENT_CHECK_LINK:
84 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<ResumeAgentCheckLink>(), return HCCL_E_PTR);
85 0 : break;
86 : // Resume 过程中接收Server借轨命令，下发给Aicpu背景线程，借轨完成通知Server
87 0 : case RETRY_RESUME_STATE_AGENT_CHANGE_LINK:
88 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<ResumeAgentChangeLink>(), return HCCL_E_PTR);
89 0 : break;
90 0 : default: {
91 0 : HCCL_ERROR("[OpRetry][Agent]CreateOpRetryAgentByState failed, state[%s] is invalid",
92 : GetReadableState(state));
93 0 : return HCCL_E_NOT_SUPPORT;
94 : }
95 : }
96 5 : retryCtx->SetRetryState(state, retryPtr);
97 5 : return HCCL_SUCCESS;
98 5 : }
99 :
100 0 : HcclResult OpRetryAgentBase::ProcessError(RetryContext* retryCtx)
101 : {
102 0 : HCCL_ERROR("[%s]OpRetryAgent run fail, rankId[%u], state[%s], IpInfo[%s]", __func__,
103 : retryCtx->rankId_, retryCtx->GetReadableCtxState(), retryCtx->GetDfxIpInfo());
104 : // 状态切换至RETRY_STATE_RESP_RUNNING_ERR（上报Server）
105 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_RESP_RUNNING_ERR, retryCtx));
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 8 : OpRetryAgentRunning::OpRetryAgentRunning()
110 : {
111 8 : lastRecvCmdTime_ = std::chrono::steady_clock::now();
112 8 : lastPollAicpuTime_ = lastRecvCmdTime_;
113 8 : pollTimeout_ = std::chrono::seconds(OP_RETRY_POLL_AICPU_ERROR_INTERVAL); // 轮询aicpu间隔
114 8 : keepTimeout_ = std::chrono::seconds(OP_RETRY_KEEP_INTERVAL); // 发送保活数据间隔
115 :
116 8 : lastPollRcTime_ = lastRecvCmdTime_;
117 8 : lastKeepTime_ = lastRecvCmdTime_;
118 8 : pollRcTimeout_ = std::chrono::seconds(OP_RETRY_POLL_RDMA_ERROR_INTERVAL); // 轮询rdma cqe间隔
119 8 : }
120 :
121 : // RETRY_STATE_AGENT_RUNNING
122 1 : HcclResult OpRetryAgentRunning::ProcessEvent(RetryContext* retryCtx)
123 : {
124 1 : HcclResult ret = HCCL_SUCCESS;
125 1 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
126 :
127 : // 定期轮询aicpu状态
128 1 : const auto pollTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - lastPollAicpuTime_);
129 1 : if (pollTime > pollTimeout_) {
130 0 : if (retryCtx->isEnableSdmaRetry_) {
131 0 : RetryState nextState = RETRY_STATE_RESERVED;
132 0 : CHK_RET(ParseKfcErr(retryCtx, nextState));
133 0 : if (nextState != RETRY_STATE_RESERVED) {
134 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_RESP_AICPU_ERR, retryCtx));
135 0 : return HCCL_SUCCESS;
136 : }
137 : }
138 0 : lastPollAicpuTime_ = curTime;
139 : }
140 :
141 : // 定期轮询 Rdma Cqe 状态
142 1 : const auto pollRcTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - lastPollRcTime_);
143 1 : if (pollRcTime > pollRcTimeout_) {
144 0 : HCCL_DEBUG("[OpRetry][Agent] OpRetryAgentRunning poll rdma err");
145 0 : if (retryCtx->isEnableBackupLink_) {
146 0 : RetryState nextState = RETRY_STATE_RESERVED;
147 : // 遍历 RDMA CQE Error 状态
148 0 : CHK_RET(ParseRdmaErr(retryCtx, nextState));
149 0 : if (nextState != RETRY_STATE_RESERVED) {
150 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
151 0 : return HCCL_SUCCESS;
152 : }
153 : }
154 0 : lastPollRcTime_ = curTime;
155 : }
156 :
157 : // OpRetryAgent Running状态下,读取到KfcStatus:kPlanSwitch或kSwitchError,切换状态机状态。
158 1 : const auto activeTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - lastPollAicpuTime_);
159 1 : KfcExecStatus &opInfo = retryCtx->localRetryInfo_.opInfo;
160 1 : CHK_RET(GetOpExecInfo(retryCtx->GetD2hPtr(), opInfo));
161 1 : const KfcStatus &aicpuState = opInfo.execStatus.kfcStatus;
162 1 : if (aicpuState == KfcStatus::kPlanSwitch || aicpuState == KfcStatus::kSwitchError) {
163 1 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_SEND_SWITCH_INFO, retryCtx));
164 1 : return HCCL_SUCCESS;
165 : }
166 :
167 : // 查看是否收到server的command
168 0 : RetryCommandInfo commandinfo;
169 0 : ret = WaitCommandWithOpId(retryCtx->agentSocket_, commandinfo);
170 0 : if (ret == HCCL_SUCCESS) {
171 0 : if (commandinfo.command == RETRY_CMD_STOP_AICPU) { // 接收到有效command信息
172 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentRunning recv command[%s] success, "
173 : "tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d], streamId[%u]",
174 : GetReadableCmd(commandinfo.command), commandinfo.opId.tag, commandinfo.opId.index,
175 : commandinfo.opId.srcRank, commandinfo.opId.detRank, commandinfo.opId.isSendRecv,
176 : commandinfo.opId.streamId);
177 0 : CHK_RET(SetOpExecCmdWithOpId(retryCtx->GetH2dPtr(), KfcCommand::kStopLaunch, commandinfo.opId));
178 0 : retryCtx->curFaultOpId = commandinfo.opId;
179 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_POLL_AICPU_STOPED, retryCtx));
180 0 : return HCCL_SUCCESS;
181 0 : } else if (commandinfo.command == RETRY_CMD_RUNNING) {
182 : // 接收到RUN命令时发送保活数据
183 0 : const auto keepTime = std::chrono::duration_cast<std::chrono::seconds>(curTime - lastKeepTime_);
184 0 : if (keepTime > keepTimeout_) {
185 0 : CHK_RET(GetRetryInfo(retryCtx, retryCtx->localRetryInfo_));
186 0 : HcclResult ret = IssueResponse(retryCtx->agentSocket_, retryCtx->localRetryInfo_);
187 0 : if (ret != HCCL_SUCCESS) { // 发送保活数据失败, 打印warning
188 0 : HCCL_WARNING("[OpRetry][Agent]OpRetryAgentRunning issue response fail, ret[%d]", ret);
189 : }
190 0 : HCCL_RUN_INFO("[OpRetry][Agent]upload tag[%s]", retryCtx->localRetryInfo_.opInfo.opId.tag);
191 0 : lastKeepTime_ = curTime;
192 : }
193 : }
194 : }
195 :
196 : // 轮询间隔
197 0 : SaluSleep(OP_RETRY_RUNNING_POLL_INTERVAL);
198 0 : return HCCL_SUCCESS;
199 : }
200 :
201 0 : HcclResult OpRetryAgentRunning::ParseRdmaErr(RetryContext* retryCtx, RetryState &nextState)
202 : {
203 0 : nextState = RETRY_STATE_RESERVED;
204 0 : CHK_RET(GetRetryInfo(retryCtx, retryCtx->localRetryInfo_));
205 0 : auto &opId = retryCtx->localRetryInfo_.opInfo.opId;
206 :
207 : //先判断是否有遗留的cqe err需要处理，没有再去心跳获取
208 0 : bool isBatchSendRecv = (opId.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV);
209 0 : bool isSendRecv = (opId.opType == HcclCMDType::HCCL_CMD_SEND)||(opId.opType == HcclCMDType::HCCL_CMD_RECEIVE);
210 0 : bool IsSupportRdmaRetry = false;
211 0 : if (isBatchSendRecv){
212 0 : if (retryCtx->isBSRRdmaSendError_){
213 0 : CHK_RET(GetBsrOpId(retryCtx, HcclSendRecvType::HCCL_SEND));
214 0 : nextState = RETRY_STATE_RESP_AICPU_ERR;
215 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcError = KfcError::kRdma;
216 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcStatus = KfcStatus::kStoplaunch;
217 0 : HCCL_RUN_INFO("[OpRetry][Agent]batchsendrecv rdma send op need retry, tag[%s] index[%u]",
218 : opId.tag, opId.index);
219 0 : return HCCL_SUCCESS;
220 : }
221 0 : if (retryCtx->isBSRRdmaRecvError_){
222 0 : CHK_RET(GetBsrOpId(retryCtx, HcclSendRecvType::HCCL_RECV));
223 0 : nextState = RETRY_STATE_RESP_AICPU_ERR;
224 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcError = KfcError::kRdma;
225 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcStatus = KfcStatus::kStoplaunch;
226 0 : HCCL_RUN_INFO("[OpRetry][Agent]batchsendrecv rdma recv op need retry, tag[%s] index[%u]",
227 : opId.tag, opId.index);
228 0 : return HCCL_SUCCESS;
229 : }
230 : }
231 : // 1. 获取 Rdma Err 信息
232 0 : std::set<std::tuple<u32, u32, u32>> infoSet;
233 0 : Heartbeat::GetInstance(retryCtx->deviceLogicId_).GetQpnErr(retryCtx->group_, infoSet);
234 0 : bool isExistQPErr = (infoSet.size() > 0);
235 0 : if (!isExistQPErr) {
236 0 : return HCCL_SUCCESS;
237 : }
238 0 : if (isBatchSendRecv) {
239 0 : for (auto &info : infoSet) {
240 0 : u32 qpn = std::get<2>(info);
241 0 : u32 qpnStatus = std::get<1>(info);
242 0 : if (qpn == opId.bsrInfo[HCCL_SEND].tpQpn && qpnStatus == RDMA_CQE_ERR_STATUS) { //SendQpn QpnStatus
243 0 : retryCtx->isBSRRdmaSendError_ = true;
244 0 : CHK_RET(SetBsrOpId(retryCtx, HcclSendRecvType::HCCL_SEND));
245 : }
246 0 : if (qpn == opId.bsrInfo[HCCL_RECV].tpQpn && qpnStatus == RDMA_CQE_ERR_STATUS) { //RecvQpn QpnStatus
247 0 : retryCtx->isBSRRdmaRecvError_ = true;
248 0 : CHK_RET(SetBsrOpId(retryCtx, HcclSendRecvType::HCCL_RECV));
249 : }
250 0 : HCCL_RUN_INFO("[OpRetry][Agent]pollcqeErr, ErrQpn [%u] SendQpn [%u] RecvQpn [%u]",
251 : qpn, opId.bsrInfo[HCCL_SEND].tpQpn, opId.bsrInfo[HCCL_RECV].tpQpn);
252 : }
253 : // 处理bsr重执行，若send/recv同时报错，优先处理send报错，重执行成功后再处理recv报错
254 : // 故障是从host侧识别出来的，然后上报故障的时候需要将aicpu侧上报的batchsendrecv刷成sendrecv
255 0 : if (retryCtx->isBSRRdmaSendError_) {
256 0 : opId.index = opId.bsrInfo[HCCL_SEND].index;
257 0 : CHK_SAFETY_FUNC_RET(memset_s(opId.tag, sizeof(opId.tag), 0, sizeof(opId.tag)));
258 0 : CHK_SAFETY_FUNC_RET(memcpy_s(opId.tag, sizeof(opId.tag), opId.bsrInfo[HCCL_SEND].bsrTag,
259 : sizeof(opId.bsrInfo[HCCL_SEND].bsrTag)));
260 0 : opId.srcRank = opId.bsrInfo[HCCL_SEND].srcRank;
261 0 : opId.detRank = opId.bsrInfo[HCCL_SEND].detRank;
262 0 : opId.streamId = opId.bsrInfo[HCCL_SEND].streamId;
263 0 : opId.isSendRecv = true;
264 0 : opId.opType = HcclCMDType::HCCL_CMD_BATCH_SEND_RECV;
265 0 : IsSupportRdmaRetry = true;
266 0 : } else if (retryCtx->isBSRRdmaRecvError_) {
267 0 : opId.index = opId.bsrInfo[HCCL_RECV].index;
268 0 : CHK_SAFETY_FUNC_RET(memset_s(opId.tag, sizeof(opId.tag), 0, sizeof(opId.tag)));
269 0 : CHK_SAFETY_FUNC_RET(memcpy_s(opId.tag, sizeof(opId.tag), opId.bsrInfo[HCCL_RECV].bsrTag,
270 : sizeof(opId.bsrInfo[HCCL_RECV].bsrTag)));
271 0 : opId.srcRank = opId.bsrInfo[HCCL_RECV].srcRank;
272 0 : opId.detRank = opId.bsrInfo[HCCL_RECV].detRank;
273 0 : opId.streamId = opId.bsrInfo[HCCL_RECV].streamId;
274 0 : opId.isSendRecv = true;
275 0 : opId.opType = HcclCMDType::HCCL_CMD_BATCH_SEND_RECV;
276 0 : IsSupportRdmaRetry = true;
277 : }
278 0 : } else if (isSendRecv) {
279 0 : auto detRank = retryCtx->localRetryInfo_.rankId == retryCtx->localRetryInfo_.opInfo.opId.detRank ?
280 : retryCtx->localRetryInfo_.opInfo.opId.srcRank : retryCtx->localRetryInfo_.opInfo.opId.detRank;
281 0 : HCCL_INFO("[OpRetry][Agent][Rdma]now in isSendRecv branch (isSendRecv[%d])", isSendRecv);
282 0 : bool isFindDstRank = false;
283 0 : for (auto &info : infoSet) {
284 0 : u32 remoteRank = std::get<0>(info);
285 0 : u32 qpnStatus = std::get<1>(info);
286 0 : if (remoteRank == detRank) {
287 0 : isFindDstRank = true;
288 0 : if (qpnStatus == RDMA_CQE_ERR_STATUS) {
289 0 : HCCL_INFO("[OpRetry][Agent][Rdma]SendRecv can support Rdma Retry");
290 0 : IsSupportRdmaRetry = true;
291 : }
292 0 : break;
293 : }
294 : }
295 0 : if(!isFindDstRank) {
296 0 : HCCL_ERROR("[OpRetry][Agent] dstRank[%u] is not in infolist, do nothing", detRank);
297 0 : nextState = RETRY_STATE_AGENT_RETRY_FAIL;
298 : }
299 0 : HCCL_INFO("[OpRetry][Agent][Rdma]SendRecv link IsSupportRdmaRetry[%d]", IsSupportRdmaRetry);
300 : } else {
301 : // 非点对点通信分支
302 0 : HCCL_INFO("[OpRetry][Agent][Rdma]now in Full link branch (isSendRecv[%d])",isSendRecv);
303 0 : for (auto &info : infoSet) {
304 0 : u32 remoteRank = std::get<0>(info);
305 0 : u32 qpnStatus = std::get<1>(info);
306 0 : HCCL_INFO("remoteRank = [%u] , status = [%u]", remoteRank, qpnStatus);
307 0 : IsSupportRdmaRetry = true; // 默认设置为支持Rdma重执行
308 0 : if (qpnStatus != RDMA_CQE_ERR_STATUS) {
309 0 : IsSupportRdmaRetry = false;
310 0 : nextState = RETRY_STATE_AGENT_RETRY_FAIL;
311 0 : HCCL_ERROR("[OpRetry][Agent] remoteRank[%u] status[%u] is not 12", remoteRank, qpnStatus);
312 0 : break;
313 : }
314 : }
315 0 : HCCL_INFO("[OpRetry][Agent][Rdma]Full link IsSupportRdmaRetry[%d]", IsSupportRdmaRetry);
316 : }
317 0 : if (IsSupportRdmaRetry) {
318 0 : nextState = RETRY_STATE_RESP_AICPU_ERR;
319 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcError = KfcError::kRdma;
320 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcStatus = KfcStatus::kStoplaunch;
321 : }
322 0 : HCCL_INFO("[OpRetry][Agent][Rdma]nextState is [%s]", GetReadableState(nextState));
323 0 : return HCCL_SUCCESS;
324 0 : }
325 :
326 0 : HcclResult OpRetryAgentRunning::ParseKfcErr(RetryContext* retryCtx, RetryState &nextState)
327 : {
328 0 : nextState = RETRY_STATE_RESERVED;
329 : // 记录上一次轮询获取的错误码, 避免日志刷屏
330 0 : KfcError lastError = retryCtx->localRetryInfo_.opInfo.execStatus.kfcError;
331 0 : CHK_RET(GetRetryInfo(retryCtx, retryCtx->localRetryInfo_));
332 0 : KfcError kfcError = retryCtx->localRetryInfo_.opInfo.execStatus.kfcError;
333 0 : uint32_t retryCnt = retryCtx->localRetryInfo_.opInfo.execStatus.retryInfo.retryCount;
334 0 : switch (kfcError) {
335 0 : case KfcError::kNone: {
336 0 : break;
337 : }
338 0 : case KfcError::kSdma: {
339 0 : HCCL_RUN_INFO("[OpRetry][Agent]Get ErrorCode[%d] rertryCnt[%u]", kfcError, retryCnt);
340 0 : nextState = RETRY_STATE_RESP_AICPU_ERR;
341 0 : break;
342 : }
343 0 : default: {
344 0 : if (lastError != kfcError) {
345 0 : HCCL_RUN_INFO("[OpRetry][Agent]KfcError[%d] is not support, do nothing", kfcError);
346 : }
347 0 : break;
348 : }
349 : }
350 0 : return HCCL_SUCCESS;
351 : }
352 :
353 : // 向server状态机发送信息
354 0 : HcclResult OpRetryAgentResponse::ProcessEvent(RetryContext* retryCtx)
355 : {
356 : // 获取预期的下一个状态
357 0 : RetryState nextState = RETRY_STATE_RESERVED;
358 0 : auto it = RETRY_AGENT_RESP_STATE_LABEL.find(retryCtx->localRetryInfo_.retryState);
359 0 : CHK_PRT_RET(it == RETRY_AGENT_RESP_STATE_LABEL.end(),
360 : HCCL_ERROR("[OpRetry][Agent]OpRetryAgentResponse fail, state[%s] is not in RETRY_AGENT_RESP_STATE_LABEL",
361 : GetReadableState(retryCtx->localRetryInfo_.retryState)), HCCL_E_INTERNAL);
362 0 : nextState = it->second;
363 0 : auto &opInfo = retryCtx->localRetryInfo_.opInfo;
364 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentResponse tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d],"\
365 : "opExeState[%d], errorCode[%d], retryCount[%u], streamId[%u], isNeedReportOpRetryErr[%d]",
366 : opInfo.opId.tag, opInfo.opId.index, opInfo.opId.srcRank, opInfo.opId.detRank, opInfo.opId.isSendRecv,
367 : opInfo.execStatus.kfcStatus, opInfo.execStatus.kfcError, opInfo.execStatus.retryInfo.retryCount,
368 : opInfo.opId.streamId, retryCtx->localRetryInfo_.isNeedReportOpRetryErr);
369 :
370 : // 发送数据
371 0 : HcclResult ret = IssueResponse(retryCtx->agentSocket_, retryCtx->localRetryInfo_);
372 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][Agent]OpRetryAgentResponse IssueResponse fail"), ret);
373 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
374 0 : return HCCL_SUCCESS;
375 : }
376 :
377 : // 从server状态机接收命令
378 0 : HcclResult OpRetryAgentWaitCmd::ProcessEvent(RetryContext* retryCtx)
379 : {
380 0 : RetryState curState = retryCtx->localRetryInfo_.retryState;
381 0 : RetryState nextState = RETRY_STATE_RESERVED;
382 0 : RetryCommandInfo commandinfo;
383 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
384 0 : const u32 timeoutValue = std::max(static_cast<u32>(GetExternalInputHcclLinkTimeOut()), OP_RETRY_SEND_RECV_TIMEOUT) + OP_RETRY_WAIT_AICPU_TIMEOUT;
385 0 : const std::chrono::seconds timeout = std::chrono::seconds(timeoutValue);
386 :
387 : // 接收到命令和当前状态不匹配时, 不做处理, 等待下一个命令, 直到命令正确或者超时
388 : while (true) {
389 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
390 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
391 0 : CHK_PRT_RET(elapsed > timeout, HCCL_ERROR("[OpRetry]WaitCommand timeout"), HCCL_E_TIMEOUT);
392 0 : CHK_PRT_RET(retryCtx->isAgentStateWaitResume_,
393 : HCCL_RUN_INFO("[OpRetry][Agent]switched state form wait cmd to wait resume"),
394 : HCCL_SUCCESS);
395 :
396 0 : HcclResult ret = WaitCommandWithOpId(retryCtx->agentSocket_, commandinfo);
397 0 : if (ret == HCCL_SUCCESS) {
398 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentGetCmd state[%s] command[%s]"\
399 : "tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d], streamid[%u]",
400 : GetReadableState(curState), GetReadableCmd(commandinfo.command),
401 : commandinfo.opId.tag, commandinfo.opId.index, commandinfo.opId.srcRank, commandinfo.opId.detRank,
402 : commandinfo.opId.isSendRecv, commandinfo.opId.streamId);
403 0 : CHK_PRT(ParseCommandWithOpId(retryCtx, commandinfo, nextState));
404 0 : if (nextState != RETRY_STATE_RESERVED) { // 接收到的命令有效
405 0 : break;
406 : }
407 : }
408 0 : }
409 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
410 0 : return HCCL_SUCCESS;
411 : }
412 :
413 0 : HcclResult OpRetryAgentWaitCmd::ParseCommandWithOpId(RetryContext* retryCtx, RetryCommandInfo &commandinfo,
414 : RetryState &nextState)
415 : {
416 0 : if (!retryCtx->isChangeLinkInfoInit_) {
417 0 : CHK_RET(InitChangeLinkInfo(retryCtx));
418 0 : retryCtx->isChangeLinkInfoInit_ = true;
419 : } else {
420 : // 增量建链场景
421 0 : CHK_RET(InitChangeLinkInfo(retryCtx, true));
422 : }
423 0 : RetryState curState = retryCtx->localRetryInfo_.retryState;
424 0 : switch (commandinfo.command) {
425 0 : case RETRY_CMD_STOP_AICPU:
426 0 : if (curState == RETRY_STATE_WAIT_CMD_STOP_AICPU) {
427 0 : CHK_RET(SetOpExecCmdWithOpId(retryCtx->GetH2dPtr(), KfcCommand::kStopLaunch, commandinfo.opId));
428 0 : retryCtx->curFaultOpId = commandinfo.opId;
429 0 : nextState = RETRY_STATE_POLL_AICPU_STOPED;
430 : }
431 0 : break;
432 0 : case RETRY_CMD_STOP_STREAM:
433 0 : if (curState == RETRY_STATE_WAIT_CMD_STOP_STREAM) {
434 0 : CHK_RET(ClearStreamWithOpId(retryCtx->opStreamPtr_, HcclRtStreamClearStep::HCCL_STREAM_STOP, commandinfo.opId,
435 : retryCtx->localRetryInfo_.opInfo.opId));
436 0 : CHK_RET(SetOpExecCmdWithOpId(retryCtx->GetH2dPtr(), KfcCommand::kStopExec, commandinfo.opId));
437 0 : nextState = RETRY_STATE_POLL_STREAM_STOPED;
438 : }
439 0 : break;
440 0 : case RETRY_CMD_CLEAR_STREAM:
441 0 : if (curState == RETRY_STATE_WAIT_CMD_CLEAR_STREAM) {
442 0 : CHK_RET(ClearStreamWithOpId(retryCtx->opStreamPtr_, HcclRtStreamClearStep::HCCL_STREAM_CLEAR, commandinfo.opId,
443 : retryCtx->localRetryInfo_.opInfo.opId));
444 0 : nextState = RETRY_STATE_RESP_STREAM_CLEARED;
445 : }
446 0 : break;
447 0 : case RETRY_CMD_STOP_TRANSPORT:
448 0 : if (curState == RETRY_STATE_WAIT_CMD_STOP_TRANSPORT) {
449 0 : CHK_RET(SetTransportStatusForStop(retryCtx));
450 0 : nextState = RETRY_STATE_RESP_STOP_TRANSPORT;
451 : }
452 0 : break;
453 0 : case RETRY_CMD_CHECK_LINK:
454 0 : if (curState == RETRY_STATE_WAIT_CMD_CHECK_LINK) {
455 0 : u32 &retryCnt = retryCtx->localRetryInfo_.opInfo.execStatus.retryInfo.retryCount;
456 0 : CommConfiger& commConfiger = CommConfiger::GetInstance();
457 0 : u32 waitTime = (retryCnt == 0) ? commConfiger.GetCommConfigRetryHoldTime(retryCtx->group_) :
458 0 : commConfiger.GetCommConfigRetryIntervalTime(retryCtx->group_);
459 0 : constexpr u32 TIME_MS_TO_US = 1000;
460 0 : SaluSleep(waitTime * TIME_MS_TO_US);
461 0 : HCCL_RUN_INFO("[OpRetry][Agent]wait for [%u]ms until the link recovers", waitTime);
462 0 : CHK_RET(GetLinkPortStatus(retryCtx, retryCtx->linkPortStatus_));
463 0 : nextState = RETRY_STATE_RESP_LINK_CHECKED;
464 : }
465 0 : break;
466 0 : case RETRY_CMD_RESUME_TRANSPORT:
467 0 : if (curState == RETRY_STATE_WAIT_CMD_RESUME_TRANSPORT) {
468 0 : CHK_RET(SetTransportStatusForResume(retryCtx));
469 : // 重新建链后给aicpu下发切换链路命令
470 0 : CHK_RET(SetOpChangeLinkInfo(retryCtx->GetH2dPtr(), KfcCommand::kChangeLink, retryCtx->localChangeLinkInfo_));
471 0 : nextState = RETRY_STATE_POLL_AICPU_CHANGED;
472 : }
473 0 : break;
474 0 : case RETRY_CMD_RESET_NOTIFY:
475 0 : if (curState == RETRY_STATE_WAIT_CMD_RESET_NOTIFY) {
476 0 : CHK_RET(ResetNotify(retryCtx));
477 0 : nextState = RETRY_STATE_RESP_NOTIFY_RESETED;
478 : }
479 0 : break;
480 0 : case RETRY_CMD_CHECK_OPNAME:
481 0 : if (curState == RETRY_STATE_WAIT_CMD_CHECK) {
482 0 : CHK_RET(GetRetryInfo(retryCtx, retryCtx->localRetryInfo_));
483 0 : nextState = RETRY_STATE_RESP_CHECK_INFO;
484 : }
485 0 : break;
486 0 : case RETRY_CMD_CAN_RETRY:
487 0 : if (curState == RETRY_STATE_WAIT_CMD_CAN_RETRY) {
488 0 : bool isSendRecv = HcclCMDType::HCCL_CMD_SEND == retryCtx->localRetryInfo_.opInfo.opId.opType ||
489 0 : HcclCMDType::HCCL_CMD_RECEIVE == retryCtx->localRetryInfo_.opInfo.opId.opType;
490 0 : u32 dstRank = retryCtx->localRetryInfo_.rankId == retryCtx->localRetryInfo_.opInfo.opId.detRank ?
491 : retryCtx->localRetryInfo_.opInfo.opId.srcRank : retryCtx->localRetryInfo_.opInfo.opId.detRank;
492 0 : if (isSendRecv) {
493 0 : Heartbeat::GetInstance(retryCtx->deviceLogicId_).ClearCqeErr(retryCtx->group_, dstRank);
494 0 : } else if (HcclCMDType::HCCL_CMD_BATCH_SEND_RECV == retryCtx->localRetryInfo_.opInfo.opId.opType){
495 0 : ResetBatchSendRecvRdmaErr(retryCtx, dstRank);
496 : } else {
497 0 : Heartbeat::GetInstance(retryCtx->deviceLogicId_).ClearAllCqeErr(retryCtx->group_);
498 : }
499 0 : CHK_RET(SetOpExecCmdWithOpId(retryCtx->GetH2dPtr(), KfcCommand::kRetry, commandinfo.opId));
500 0 : nextState = RETRY_STATE_POLL_AICPU_RETRYEND;
501 : }
502 0 : break;
503 0 : case RETRY_CMD_RETRY_CONSTRAINT_FAIL:
504 0 : nextState = RETRY_STATE_AGENT_RETRY_FAIL;
505 0 : retryCtx->localRetryInfo_.isNeedReportOpRetryErr = true;
506 0 : HCCL_RUN_INFO("[OpRetry][Agent]Retry is constraint(OpName is inconsistent or Inplace Error)");
507 0 : break;
508 0 : case RETRY_CMD_RETRY_FAIL:
509 0 : nextState = RETRY_STATE_AGENT_RETRY_FAIL;
510 0 : break;
511 0 : default: { // 命令非当前状态预期, 不处理
512 0 : break;
513 : }
514 : }
515 0 : return HCCL_SUCCESS;
516 : }
517 :
518 0 : void OpRetryAgentWaitCmd::ResetBatchSendRecvRdmaErr(RetryContext* retryCtx, u32 dstRank)
519 : {
520 0 : bool isBatchSendRecv = (retryCtx->localRetryInfo_.opInfo.opId.opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV);
521 0 : if (isBatchSendRecv) {
522 0 : auto curOpIdTag = std::string(reinterpret_cast<const char*>(retryCtx->localRetryInfo_.opInfo.opId.tag));
523 0 : auto curOpIdindex = retryCtx->localRetryInfo_.opInfo.opId.index;
524 0 : auto remainSendOpIdTag = std::string(reinterpret_cast<const char*>(retryCtx->RemainSendOpId_.tag));
525 0 : auto remainRecvOpIdTag = std::string(reinterpret_cast<const char*>(retryCtx->RemainRecvOpId_.tag));
526 0 : if (curOpIdTag == remainSendOpIdTag && curOpIdindex == retryCtx->RemainSendOpId_.index) {
527 0 : retryCtx->isBSRRdmaSendError_ = false;
528 0 : HCCL_INFO("[OpRetry][Agent] bsr send clear cqe err, remoterank[%u], qpn[%u]",
529 : dstRank, retryCtx->RemainSendOpId_.bsrInfo[HCCL_SEND].tpQpn);
530 0 : Heartbeat::GetInstance(retryCtx->deviceLogicId_).ClearCqeErr(retryCtx->group_, dstRank,
531 : retryCtx->RemainSendOpId_.bsrInfo[HCCL_SEND].tpQpn);
532 : }
533 0 : if (curOpIdTag == remainRecvOpIdTag && curOpIdindex == retryCtx->RemainRecvOpId_.index){
534 0 : retryCtx->isBSRRdmaRecvError_ = false;
535 0 : HCCL_INFO("[OpRetry][Agent] bsr recv clear cqe err, remoterank[%u], qpn[%u]",
536 : dstRank, retryCtx->RemainRecvOpId_.bsrInfo[HCCL_RECV].tpQpn);
537 0 : Heartbeat::GetInstance(retryCtx->deviceLogicId_).ClearCqeErr(retryCtx->group_, dstRank,
538 : retryCtx->RemainRecvOpId_.bsrInfo[HCCL_RECV].tpQpn);
539 : }
540 0 : }
541 0 : return ;
542 : }
543 :
544 1 : HcclResult OpRetryAgentPollAicpuStop::ProcessEvent(RetryContext* retryCtx)
545 : {
546 1 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
547 1 : const std::chrono::seconds timeout = std::chrono::seconds(OP_RETRY_WAIT_AICPU_TIMEOUT);
548 1 : RetryState curState = retryCtx->GetRetryState();
549 1 : RetryState nextState = RETRY_STATE_RESERVED;
550 : while (true) {
551 1 : CHK_PRT_RET(retryCtx->isAgentStateWaitResume_,
552 : HCCL_RUN_INFO("[OpRetry][Agent]switched state form poll aicpu to wait resume"),
553 : HCCL_SUCCESS);
554 : // 读取aicpuCtx中的状态
555 1 : KfcExecStatus &opInfo = retryCtx->localRetryInfo_.opInfo;
556 1 : CHK_RET(GetOpExecInfo(retryCtx->GetD2hPtr(), opInfo));
557 1 : const KfcStatus &aicpuState = opInfo.execStatus.kfcStatus;
558 1 : const char* tag = reinterpret_cast<const char*>(opInfo.opId.tag);
559 1 : u32 index = opInfo.opId.index;
560 1 : KfcError errorCode = opInfo.execStatus.kfcError;
561 :
562 2 : std::string curFaultTag = std::string(reinterpret_cast<const char*>(retryCtx->curFaultOpId.tag));
563 1 : std::string curd2hTag = std::string(reinterpret_cast<const char*>(opInfo.opId.tag));
564 1 : switch(curState) {
565 1 : case RETRY_STATE_POLL_AICPU_STOPED:
566 1 : if (aicpuState == KfcStatus::kStoplaunch ||
567 1 : aicpuState == KfcStatus::kStopExec) {
568 0 : if ((retryCtx->curFaultOpId.isSendRecv && curFaultTag == curd2hTag) || !opInfo.opId.isSendRecv)
569 : {
570 0 : HCCL_RUN_INFO("[OpRetry][Agent]curFaultTag[%s] curd2hTag[%s], isSendRecv[%u]",
571 : curFaultTag.c_str(), curd2hTag.c_str(), retryCtx->curFaultOpId.isSendRecv);
572 0 : nextState = RETRY_STATE_RESP_AICPU_STOPED;
573 : }
574 1 : } else if (aicpuState == KfcStatus::kRetryError && errorCode == KfcError::kExecConstraint) {
575 1 : retryCtx->localRetryInfo_.isNeedReportOpRetryErr = true;
576 1 : HCCL_ERROR("[OpRetry][Agent]can not retry for constraint");
577 1 : nextState = RETRY_STATE_RESP_RUNNING_ERR;
578 1 : break;
579 : }
580 0 : break;
581 0 : case RETRY_STATE_POLL_STREAM_STOPED:
582 0 : if (aicpuState == KfcStatus::kStopExec) {
583 0 : nextState = RETRY_STATE_RESP_STREAM_STOPED;
584 0 : } else if (aicpuState == KfcStatus::kRetryError && errorCode == KfcError::kExecConstraint) {
585 0 : retryCtx->localRetryInfo_.isNeedReportOpRetryErr = true;
586 0 : HCCL_ERROR("[OpRetry][Agent]can not retry for constraint");
587 0 : nextState = RETRY_STATE_RESP_RUNNING_ERR;
588 0 : break;
589 : }
590 0 : break;
591 0 : case RETRY_STATE_POLL_AICPU_CHANGED:
592 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentPollAicpuStop hostState[%s], aicpuState[%d]",
593 : GetReadableState(curState), aicpuState);
594 0 : if (aicpuState == KfcStatus::kChanged) {
595 0 : nextState = RETRY_STATE_RESP_RESUME_TRANSPORT;
596 : }
597 0 : break;
598 0 : case RETRY_STATE_POLL_AICPU_RETRYEND:
599 0 : if (aicpuState == KfcStatus::kStoplaunch || aicpuState == KfcStatus::kRuning ||
600 0 : aicpuState == KfcStatus::kEnd){
601 0 : nextState = RETRY_STATE_RESP_AICPU_RETRYEND;
602 : }
603 0 : break;
604 0 : default: {
605 0 : HCCL_ERROR("[OpRetry][Agent]OpRetryAgentPollAicpuStop state[%s] is invalid",
606 : GetReadableState(curState));
607 0 : return HCCL_E_INTERNAL;
608 : }
609 : }
610 :
611 : // 执行成功, 跳出循环进入下一个状态
612 1 : if (nextState != RETRY_STATE_RESERVED) {
613 1 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentPollAicpuStop success, retryState[%s], aicpuState[%d], "\
614 : "tag[%s], index[%u]", GetReadableState(curState), aicpuState, tag, index);
615 1 : HCCL_RUN_INFO("[OpRetry][agent pollaicpu OpId]tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d],"
616 : "streamid[%u], retryCnt[%u]",
617 : opInfo.opId.tag, opInfo.opId.index, opInfo.opId.srcRank, opInfo.opId.detRank,
618 : opInfo.opId.isSendRecv, opInfo.opId.streamId, opInfo.execStatus.retryInfo.retryCount);
619 1 : break;
620 : }
621 :
622 : // 超时机制
623 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
624 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
625 0 : CHK_PRT_BREAK(elapsed >= timeout,
626 : HCCL_ERROR("[OpRetry][Agent]OpRetryAgentPollAicpuStop timeout, retryState[%s], aicpuState[%d], "\
627 : "tag[%s], index[%u]", GetReadableState(curState), aicpuState, tag, index),
628 : nextState = RETRY_STATE_RESP_RUNNING_ERR);
629 :
630 : // 轮询间隔
631 0 : SaluSleep(OP_RETRY_POLL_AICPU_STATE_INTERVAL);
632 2 : }
633 :
634 1 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
635 1 : return HCCL_SUCCESS;
636 : }
637 :
638 : // 向server状态机发送主备链路信息
639 0 : HcclResult OpRetryAgentResponseLinkInfo::ProcessEvent(RetryContext* retryCtx)
640 : {
641 0 : HCCL_INFO("[OpRetry][Agent]OpRetryAgentResponseLinkInfo begin");
642 : // 获取预期的下一个状态
643 0 : RetryState nextState = RETRY_STATE_WAIT_CHANGE_LINK_INFO;
644 :
645 : // 发送数据
646 0 : HcclResult ret = IssueLinkPortCheckResult(retryCtx->agentSocket_, retryCtx->linkPortStatus_);
647 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][Agent]OpRetryAgentResponseLinkInfo IssueResponse fail"), ret);
648 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
649 0 : HCCL_INFO("[OpRetry][Agent]OpRetryAgentResponseLinkInfo success");
650 0 : return HCCL_SUCCESS;
651 : }
652 :
653 0 : void OpRetryAgentWaitChangeLinkInfo::UpdateChangeLinkInfo(ChangeLinkInfo &localChangeLinkInfo, ChangeLinkInfo &recvChangeLinkInfo)
654 : {
655 : // 记录localChangeLinkInfo_中已有的数据
656 0 : std::unordered_map<u32, u32> remoteRankPosition; // {remoteRank: position}
657 0 : for (u32 i = 0; i < localChangeLinkInfo.remoteRankNum; i++) {
658 0 : remoteRankPosition.insert({localChangeLinkInfo.remoteRankList[i], i});
659 : }
660 : // 将接收到的recvChangeLinkInfo更新到localChangeLinkInfo_中
661 0 : for (u32 i = 0; i < recvChangeLinkInfo.remoteRankNum; i++) {
662 0 : u32 remoteRank = recvChangeLinkInfo.remoteRankList[i];
663 0 : bool isUseDefaultPort = recvChangeLinkInfo.isUseDefaultPort[i];
664 0 : if (remoteRankPosition.find(remoteRank) != remoteRankPosition.end()) {
665 : // 若remoteRank在localChangeLinkInfo_中，更新其端口使用情况
666 0 : localChangeLinkInfo.isUseDefaultPort[remoteRankPosition[remoteRank]] = isUseDefaultPort;
667 0 : HCCL_RUN_INFO("[OpRetry][Agent]update remoteRank[%u] to isUseDefaultPort[%d]", remoteRank, isUseDefaultPort);
668 : } else {
669 : // 若remoteRank不在localChangeLinkInfo_中，则添加到localChangeLinkInfo_中
670 0 : u32 position = localChangeLinkInfo.remoteRankNum;
671 0 : localChangeLinkInfo.remoteRankList[position] = remoteRank;
672 0 : localChangeLinkInfo.isUseDefaultPort[position] = isUseDefaultPort;
673 0 : localChangeLinkInfo.remoteRankNum += 1;
674 0 : HCCL_RUN_INFO("[OpRetry][Agent]insert remoteRank[%u] to isUseDefaultPort[%d]", remoteRank, isUseDefaultPort);
675 : }
676 : }
677 0 : return ;
678 0 : }
679 :
680 : // 从server状态机接收主备借轨命令
681 0 : HcclResult OpRetryAgentWaitChangeLinkInfo::ProcessEvent(RetryContext* retryCtx)
682 : {
683 0 : HCCL_INFO("[OpRetry][Agent]OpRetryAgentWaitChangeLinkInfo begin");
684 0 : RetryState nextState = RETRY_STATE_RESERVED;
685 :
686 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
687 0 : const u32 timeoutValue = std::max(static_cast<u32>(GetExternalInputHcclLinkTimeOut()), OP_RETRY_SEND_RECV_TIMEOUT) + OP_RETRY_WAIT_AICPU_TIMEOUT;
688 0 : const std::chrono::seconds timeout = std::chrono::seconds(timeoutValue);
689 0 : ChangeLinkInfo tmpRecvChangeLinkInfo;
690 : // 接收到命令和当前状态不匹配时, 不做处理, 等待下一个命令, 直到命令正确或者超时
691 : while (true) {
692 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
693 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
694 0 : CHK_PRT_RET(elapsed > timeout, HCCL_ERROR("[OpRetry]WaitChangeLink timeout"), HCCL_E_TIMEOUT);
695 0 : CHK_PRT_RET(retryCtx->isAgentStateWaitResume_,
696 : HCCL_RUN_INFO("[OpRetry][Agent]switched state form wait change link to wait resume"),
697 : HCCL_SUCCESS);
698 :
699 0 : HcclResult ret = WaitChangeLink(retryCtx->agentSocket_, tmpRecvChangeLinkInfo);
700 0 : if (ret == HCCL_SUCCESS) {
701 0 : HCCL_INFO("[OpRetry][Agent]WaitChangeLink success");
702 : // 将接收到的ChangeLinkInfo更新到已有的changeLinkInfo中
703 0 : UpdateChangeLinkInfo(retryCtx->localChangeLinkInfo_, tmpRecvChangeLinkInfo);
704 : // agent接收的changeLinkInfo信息
705 0 : std::string changeLinkInfoStr = "agent:";
706 0 : for (u32 i = 0; i < retryCtx->localChangeLinkInfo_.remoteRankNum; i++) {
707 0 : changeLinkInfoStr += (std::to_string(retryCtx->localChangeLinkInfo_.remoteRankList[i]) + ":" +
708 0 : std::to_string(retryCtx->localChangeLinkInfo_.isUseDefaultPort[i]) + "; ");
709 : }
710 0 : HCCL_RUN_INFO("[OpRetry][Agnet]changeLinkInfoStr:%s", changeLinkInfoStr.c_str());
711 :
712 : // 收到changelinkinfo后切换到RETRY_STATE_WAIT_CMD_RESUME_TRANSPORT状态等待接收resume transport命令
713 0 : nextState = RETRY_STATE_WAIT_CMD_RESUME_TRANSPORT;
714 0 : break;
715 0 : }
716 0 : }
717 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
718 0 : HCCL_INFO("[OpRetry][Agent]OpRetryAgentWaitChangeLinkInfo success");
719 0 : return HCCL_SUCCESS;
720 : }
721 :
722 : // RETRY_STATE_AGENT_RETRY_FAIL
723 0 : HcclResult OpRetryAgentRetryFail::ProcessEvent(RetryContext* retryCtx)
724 : {
725 0 : HCCL_INFO("[OpRetry][Agent]OpRetryAgentRetryFail, set state to running");
726 0 : if (retryCtx->localRetryInfo_.isNeedReportOpRetryErr) {
727 0 : CHK_RET(SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kReportRetryErr));
728 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentRetryFail, isNeeReportOpRetryErr[%d]", retryCtx->localRetryInfo_.isNeedReportOpRetryErr);
729 : } else{
730 0 : CHK_RET(SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kExit));
731 : }
732 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_AGENT_RUNNING, retryCtx));
733 0 : Heartbeat::GetInstance(retryCtx->deviceLogicId_).BroadcastCqeErr(retryCtx->group_);
734 0 : return HCCL_SUCCESS;
735 : }
736 :
737 0 : HcclResult OpRetryAgentWaitResume::ProcessEvent(RetryContext* retryCtx)
738 : {
739 0 : if (!retryCtx->isAgentStateWaitResume_ && !retryCtx->haveCommEnableBackupLink_) {
740 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_AGENT_RUNNING, retryCtx));
741 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentWaitResume, group[%s], no comm enable backup link, set state to running", retryCtx->group_.c_str());
742 0 : return HCCL_SUCCESS;
743 : }
744 0 : RetryCommandInfo commandInfo;
745 0 : HcclResult ret = WaitCommandWithOpId(retryCtx->agentSocket_, commandInfo);
746 0 : if(ret == HCCL_SUCCESS) {
747 0 : HCCL_INFO("[OpRetry][Agent]OpRetryAgentWaitResume, rankId[%u], command[%s], group[%s], isAgentStateWaitResume_[%d]", retryCtx->localRetryInfo_.rankId,
748 : GetReadableCmd(commandInfo.command), retryCtx->group_.c_str(), retryCtx->isAgentStateWaitResume_);
749 : }
750 0 : if (commandInfo.command == RESUME_CMD_RUNNING) {
751 0 : retryCtx->isRecivedCmdToRunning = true;
752 0 : } else if (commandInfo.command == RESUME_CMD_CHECK_LINK) {
753 0 : retryCtx->isRecivedCmdToCheckLink = true;
754 : }
755 0 : if (!retryCtx->isAgentStateWaitResume_ && retryCtx->isRecivedCmdToRunning) {
756 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_AGENT_RUNNING, retryCtx));
757 0 : retryCtx->isRecivedCmdToRunning = false;
758 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentWaitResume, set state to running");
759 0 : } else if (!retryCtx->isAgentStateWaitResume_ && retryCtx->isRecivedCmdToCheckLink) {
760 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_RESUME_STATE_AGENT_CHECK_LINK, retryCtx));
761 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentWaitResume, set state to check link");
762 : }
763 0 : return HCCL_SUCCESS;
764 : }
765 :
766 :
767 : // RETRY_STATE_WAIT_CMD_SEND_AICPU
768 2 : HcclResult SwitchNicAgentWaitCmd::ParseCommand(RetryContext* retryCtx, RetryCommand &command, RetryState &nextState)
769 : {
770 2 : switch (command) {
771 2 : case RETRY_CMD_NOTIFY_SWITCH_SUC:
772 2 : HCCL_INFO("[SwitchNic][Agent][WaitCmd] switch nic success, rank[%u]", retryCtx->rankId_);
773 : // OpRetryAgent接收到全局通信域的成功/错误信息后，刷新lastLinkPortStatus_信息
774 4 : for(u32 i = 0; i < retryCtx->switchInfo_.switchRankNum; i++) {
775 2 : if (retryCtx->switchInfo_.switchRankList[i] == retryCtx->rankId_) {
776 2 : retryCtx->isUseDefaultPort_ = !retryCtx->switchInfo_.switchUseBackup[i];
777 : }
778 : }
779 6 : for (u32 i = 0; i < retryCtx->switchInfo_.remoteRankNum; i++) {
780 4 : if (retryCtx->switchInfo_.remoteRankNicStatus[i] == CONNECT_REMOTE_DEFAULT) {
781 2 : retryCtx->lastLinkPortStatus_[i] = true;
782 : }
783 4 : if (retryCtx->switchInfo_.remoteRankNicStatus[i] == CONNECT_REMOTE_BACKUP) {
784 2 : retryCtx->lastLinkPortStatus_[i] = false;
785 : }
786 : }
787 2 : CHK_RET(SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kAllSwitched));
788 2 : nextState = RETRY_STATE_AGENT_RUNNING;
789 2 : break;
790 0 : case RETRY_CMD_NOTIFY_SWITCH_FAIL:
791 0 : HCCL_ERROR("[SwitchNic][Agent][WaitCmd] switch nic failed, rank[%u]", retryCtx->rankId_);
792 0 : CHK_RET(SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kSwitchFail));
793 0 : nextState = RETRY_STATE_AGENT_RUNNING;
794 0 : break;
795 0 : case RETRY_CMD_RUNNING:
796 0 : HCCL_DEBUG("[SwitchNic][Agent][WaitCmd] rank[%u] recv running command from server, ignored",
797 : retryCtx->rankId_);
798 0 : break;
799 0 : default: { // 命令非当前状态预期
800 0 : HCCL_ERROR("[SwitchNic][Agent][WaitCmd] rank[%u], recv unexpected parse command[%s:%u].",
801 : retryCtx->rankId_, GetReadableCmd(command), command);
802 0 : return HCCL_E_INTERNAL;
803 : }
804 : }
805 2 : return HCCL_SUCCESS;
806 : }
807 :
808 2 : HcclResult SwitchNicAgentWaitCmd::ProcessEvent(RetryContext* retryCtx)
809 : {
810 2 : HCCL_RUN_INFO("[SwitchNic][Agent] rank[%u] begin to wait server cmd, set state to wait", retryCtx->rankId_);
811 2 : RetryState curState = retryCtx->localRetryInfo_.retryState;
812 2 : RetryState nextState = RETRY_STATE_RESERVED;
813 : RetryCommand command;
814 2 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
815 2 : const auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut() * ACTIVE_SWITCH_TIMES);
816 :
817 : // 接收到命令和当前状态不匹配时, 不做处理, 等待下一个命令, 直到命令正确或者超时
818 : while (true) {
819 2 : CHK_PRT_RET(retryCtx->isAgentStateWaitResume_,
820 : HCCL_RUN_INFO("[OpRetry][Agent]switched state form switch nic to wait resume"),
821 : HCCL_SUCCESS);
822 2 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
823 2 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
824 2 : CHK_PRT_RET(elapsed > timeout,
825 : HCCL_ERROR("[SwitchNic][Agent] timeout in getting cmd from server, waitime[%u>%u]",
826 : elapsed, timeout), HCCL_E_TIMEOUT);
827 :
828 2 : HcclResult ret = WaitCommand(retryCtx->agentSocket_, command);
829 2 : if (ret == HCCL_SUCCESS) {
830 2 : HCCL_DEBUG("[SwitchNic][Agent]SwitchNicAgentWaitCmd state[%s] command[%s:%u]",
831 : GetReadableState(curState), GetReadableCmd(command), command);
832 2 : CHK_PRT(ParseCommand(retryCtx, command, nextState));
833 2 : if (nextState != RETRY_STATE_RESERVED) { // 接收到的命令有效
834 2 : break;
835 : }
836 : }
837 0 : }
838 2 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
839 2 : return HCCL_SUCCESS;
840 : }
841 :
842 1 : HcclResult SwitchNicAgentSendSwitchInfo::ChangeAicpuStatus(RetryContext* retryCtx)
843 : {
844 1 : HcclResult ret = SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kWaitSwitchNic);
845 1 : if (ret != HCCL_SUCCESS) {
846 0 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], ChangeAicpuStatus, SetOpExecCmd to waitSwitchNic fail, ",
847 : retryCtx->rankId_);
848 : }
849 1 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
850 1 : const std::chrono::seconds timeout = std::chrono::seconds(OP_RETRY_WAIT_AICPU_TIMEOUT);
851 : while (true) {
852 2486232 : CHK_PRT_RET(retryCtx->isAgentStateWaitResume_,
853 : HCCL_RUN_INFO("[OpRetry][Agent]switched state form send swhitch Nic info to wait resume"),
854 : HCCL_SUCCESS);
855 2486231 : KfcExecStatus &opInfo = retryCtx->localRetryInfo_.opInfo;
856 2486231 : CHK_RET(GetOpExecInfo(retryCtx->GetD2hPtr(), opInfo));
857 2486231 : const KfcStatus &aicpuState = opInfo.execStatus.kfcStatus;
858 2486231 : if (aicpuState == KfcStatus::kWaitSwitchRes) {
859 0 : break;
860 : }
861 :
862 2486231 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
863 2486231 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
864 2486231 : CHK_PRT_RET(elapsed >= timeout,
865 : HCCL_ERROR("[SwitchNic][Agent]SwitchNicAgentSendSwitchInfo, WaitAicpuResponse timeout, "
866 : "rank[%u], aicpuState[%d]", retryCtx->rankId_, aicpuState), HCCL_E_TIMEOUT);
867 2486230 : }
868 0 : return HCCL_SUCCESS;
869 : }
870 :
871 1 : HcclResult SwitchNicAgentSendSwitchInfo::CheckLocalPortStatus(RetryContext* retryCtx)
872 : {
873 1 : HcclResult ret = HcclNetDevGetPortStatus(retryCtx->netDevCtx_, retryCtx->linkPortStatus_.defaultPort);
874 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SwitchNic][Agent] rank[%u], get default port status fail.",
875 : retryCtx->rankId_), HCCL_E_INTERNAL);
876 1 : ret = HcclNetDevGetPortStatus(retryCtx->backUpNetDevCtx_, retryCtx->linkPortStatus_.backupPort);
877 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SwitchNic][Agent] rank[%u], get backup port status fail.",
878 : retryCtx->rankId_), HCCL_E_INTERNAL);
879 1 : return HCCL_SUCCESS;
880 : }
881 :
882 : // RETRY_STATE_SEND_SWITCH_INFO
883 1 : HcclResult SwitchNicAgentSendSwitchInfo::ProcessEvent(RetryContext* retryCtx)
884 : {
885 1 : HCCL_RUN_INFO("[SwitchNic][Agent] rank[%u] begin to send switch info, set state to switch", retryCtx->rankId_);
886 :
887 1 : ActiveSwitchInfo &switchInfo = retryCtx->switchInfo_;
888 1 : const KfcStatus transportStatus = retryCtx->localRetryInfo_.opInfo.execStatus.kfcStatus;
889 1 : switchInfo.refreshTransportFin = (transportStatus == KfcStatus::kPlanSwitch) ? true : false;
890 1 : if (!switchInfo.refreshTransportFin) {
891 0 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], recv refresh transport kfcStatus[%u], "
892 : "refreshTransportFin is set to false, ", retryCtx->rankId_, transportStatus);
893 : }
894 :
895 1 : HcclResult ret = ChangeAicpuStatus(retryCtx);
896 1 : if (ret != HCCL_SUCCESS) {
897 1 : switchInfo.refreshTransportFin = false;
898 1 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], ChangeAicpuStatus fail, refreshTransportFin is set to false.",
899 : retryCtx->rankId_);
900 : }
901 :
902 : // 校验本端准备使用的主/备网卡状态
903 1 : ret = CheckLocalPortStatus(retryCtx);
904 1 : if (ret != HCCL_SUCCESS) {
905 0 : switchInfo.localPortsCheckRet = false;
906 0 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], get local port status fail, "
907 : "localPortsCheckRet is set to false.", retryCtx->rankId_);
908 : }
909 :
910 : // 并将本端完成借轨的flag, 以及之前登记的switchRanks信息发送给OpRetryServer
911 1 : bool needCheckDefaultNic = false;
912 1 : bool needCheckBackupNic = false;
913 1 : ret = GetSwitchRanks(retryCtx, needCheckDefaultNic, needCheckBackupNic);
914 1 : if (ret != HCCL_SUCCESS) {
915 0 : switchInfo.refreshTransportFin = false;
916 0 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], get switch rank info fail, "
917 : "refreshTransportFin is set to false.", retryCtx->rankId_);
918 : }
919 :
920 1 : switchInfo.defaultPortStatus = retryCtx->linkPortStatus_.defaultPort;
921 1 : switchInfo.backupPortStatus = retryCtx->linkPortStatus_.backupPort;
922 1 : if ((needCheckDefaultNic && !switchInfo.defaultPortStatus) ||
923 0 : (needCheckBackupNic && !switchInfo.backupPortStatus)) {
924 1 : switchInfo.localPortsCheckRet = false;
925 1 : HCCL_ERROR("[SwitchNic][Agent] localPortsCheckRet is false, needCheckDefaultNic[%u], needCheckBackupNic[%u],"
926 : "localPortsCheckRet[%u], defaultPortStatus[%u], backupPortStatus[%u], rank[%u]", needCheckDefaultNic,
927 : needCheckBackupNic, switchInfo.localPortsCheckRet, switchInfo.defaultPortStatus, switchInfo.backupPortStatus,
928 : retryCtx->rankId_);
929 : } else {
930 0 : switchInfo.localPortsCheckRet = true;
931 : }
932 1 : HCCL_RUN_INFO("[SwitchNic][Agent] switch info, refresh transport kfcStatus[%u], needCheckDefaultNic[%u], "
933 : "needCheckBackupNic[%u], localPortsCheckRet[%u], defaultPortStatus[%u], backupPortStatus[%u], "
934 : "refreshTransportFin[%u], switchRankNum[%u], remoteRankNum[%u], rank[%u]",
935 : transportStatus, needCheckDefaultNic, needCheckBackupNic, switchInfo.localPortsCheckRet,
936 : switchInfo.defaultPortStatus, switchInfo.backupPortStatus, switchInfo.refreshTransportFin,
937 : switchInfo.switchRankNum, switchInfo.remoteRankNum, retryCtx->rankId_);
938 : // 第一次发送retryInfo，目的是为了server确认为主动接轨场景再接收switchInfo
939 2 : RetryInfo retryInfo = {};
940 1 : retryInfo.retryState = RETRY_STATE_SEND_SWITCH_INFO;
941 1 : retryInfo.rankId = retryCtx->rankId_;
942 1 : ret = IssueResponse(retryCtx->agentSocket_, retryInfo);
943 1 : if (ret == HCCL_SUCCESS) {
944 1 : ret = IssueActiveSwitchInfo(retryCtx->agentSocket_, switchInfo);
945 : } else {
946 0 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], issue response to server fail.", retryCtx->rankId_);
947 : }
948 1 : if (ret != HCCL_SUCCESS) {
949 0 : HCCL_ERROR("[SwitchNic][Agent] rank[%u], issue response or active switch info to server fail, "
950 : "send result to aicpu.", retryCtx->rankId_);
951 0 : CHK_RET(SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kSwitchFail));
952 0 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_AGENT_RUNNING, retryCtx));
953 : } else {
954 1 : CHK_RET(CreateOpRetryAgentByState(RETRY_STATE_WAIT_CMD_SEND_AICPU, retryCtx));
955 : }
956 1 : return HCCL_SUCCESS;
957 : }
958 :
959 0 : HcclResult ResumeAgentCheckLink::ProcessEvent(RetryContext* retryCtx)
960 : {
961 : // 快恢阶段需要获取通信域内所有使用RDMA链路的rank列表, isGetGroupAllRemoteRank为true
962 0 : if (!retryCtx->isChangeLinkInfoInit_) {
963 0 : CHK_RET(InitChangeLinkInfo(retryCtx, false, true));
964 0 : retryCtx->isChangeLinkInfoInit_ = true;
965 : } else {
966 : // BatchSendRecv算子增量建链场景
967 0 : CHK_RET(InitChangeLinkInfo(retryCtx, true, true));
968 : }
969 :
970 0 : CHK_RET(SetTransportStatusForStop(retryCtx));
971 0 : RetryState nextState = RETRY_RESUME_STATE_AGENT_CHANGE_LINK;
972 0 : HCCL_RUN_INFO("[OpRetry][Agent]OpRetryAgentWaitResume, start to check link");
973 : // 获取当前主备网口状态，并且回复Server
974 0 : CHK_RET(GetLinkPortStatus(retryCtx, retryCtx->linkPortStatus_, true));
975 0 : HcclResult ret = IssueLinkPortCheckResult(retryCtx->agentSocket_, retryCtx->linkPortStatus_);
976 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][Agent]ResumeAgentCheckLink IssueResponse fail"), ret);
977 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
978 0 : return HCCL_SUCCESS;
979 : }
980 :
981 0 : HcclResult ResumeAgentChangeLink::ProcessEvent(RetryContext *retryCtx)
982 : {
983 0 : HCCL_RUN_INFO("[OpRetry][Agent][Resume]ResumeAgentChangeLink group[%s] rank[%d] start",
984 : retryCtx->group_.c_str(), retryCtx->rankId_);
985 0 : RetryState nextState = RETRY_STATE_RESERVED;
986 : // 接收到Server下发的切换链路信息和重建transport命令
987 0 : CHK_RET(WaitResumeCmdResumeTransport(retryCtx));
988 : // 重建当前选择的链路并拷贝到device侧，给aicpu下发切换链路命令
989 0 : CHK_RET(SetTransportStatusForResume(retryCtx));
990 0 : HCCL_INFO("[OpRetry][Agent][Resume]ResumeAgentChangeLink group[%s] rank[%d] SetTransportStatusForResume finish",
991 : retryCtx->group_.c_str(), retryCtx->rankId_);
992 : // 等待aicpu切换链路完成
993 0 : CHK_RET(SetOpChangeLinkInfo(retryCtx->GetH2dPtr(), KfcCommand::NsChangeLink, retryCtx->localChangeLinkInfo_));
994 0 : HCCL_INFO("[OpRetry][Agent][Resume]ResumeAgentChangeLink group[%s] rank[%d] SetOpChangeLinkInfo finish",
995 : retryCtx->group_.c_str(), retryCtx->rankId_);
996 0 : CHK_RET(WaitAndRespLinkChanged(retryCtx, nextState));
997 0 : if (nextState != RETRY_STATE_AGENT_RUNNING) {
998 0 : HCCL_ERROR("[OpRetry][Agent]ResumeAgentChangeLink WaitAndRespLinkChanged fail, nextState[%s]", nextState);
999 : }
1000 0 : CHK_RET(CreateOpRetryAgentByState(nextState, retryCtx));
1001 0 : return HCCL_SUCCESS;
1002 : }
1003 :
1004 0 : HcclResult ResumeAgentChangeLink::WaitResumeCmdResumeTransport(RetryContext *retryCtx)
1005 : {
1006 0 : ChangeLinkInfo tmpRecvChangeLinkInfo;
1007 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
1008 0 : const u32 timeoutValue = std::max(static_cast<u32>(GetExternalInputHcclLinkTimeOut()), OP_RETRY_SEND_RECV_TIMEOUT) + OP_RETRY_WAIT_AICPU_TIMEOUT;
1009 0 : const std::chrono::seconds timeout = std::chrono::seconds(timeoutValue);
1010 : // 接收到命令和当前状态不匹配时, 不做处理, 等待下一个命令, 直到命令正确或者超时
1011 : while (true) {
1012 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
1013 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
1014 0 : CHK_PRT_RET(elapsed > timeout, HCCL_ERROR("[OpRetry][Agent]ResumeAgentChangeLink timeout"), HCCL_E_TIMEOUT);
1015 :
1016 0 : HcclResult ret = WaitChangeLink(retryCtx->agentSocket_, tmpRecvChangeLinkInfo);
1017 0 : if (ret == HCCL_SUCCESS) {
1018 0 : HCCL_INFO("[OpRetry][Agent][Resume]WaitResumeCmdResumeTransport receive changelinkinfo success, rank[%d], group[%s]",
1019 : retryCtx->rankId_, retryCtx->group_.c_str());
1020 0 : UpdateChangeLinkInfo(retryCtx->localChangeLinkInfo_, tmpRecvChangeLinkInfo);
1021 : // agent接收到的changeLinkInfo信息
1022 0 : std::string changeLinkInfoStr = "agent:";
1023 0 : for (u32 i = 0; i < retryCtx->localChangeLinkInfo_.remoteRankNum; i++) {
1024 0 : changeLinkInfoStr += (std::to_string(retryCtx->localChangeLinkInfo_.remoteRankList[i]) + ":" +
1025 0 : std::to_string(retryCtx->localChangeLinkInfo_.isUseDefaultPort[i]) + "; ");
1026 : }
1027 0 : HCCL_RUN_INFO("[OpRetry][Agent][Resume]changeLinkInfoStr:%s", changeLinkInfoStr.c_str());
1028 0 : break;
1029 0 : }
1030 0 : }
1031 0 : HCCL_INFO("[OpRetry][Agent][Resume]WaitResumeCmdResumeTransport begin to wait command to changelink, rank[%d], group[%s]",
1032 : retryCtx->rankId_, retryCtx->group_.c_str());
1033 : // 轮询等待接收借轨命令
1034 : while (true) {
1035 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
1036 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
1037 0 : CHK_PRT_RET(elapsed > timeout, HCCL_ERROR("[OpRetry]WaitResumeCmdChangeLink timeout"), HCCL_E_TIMEOUT);
1038 0 : RetryCommandInfo commandInfo;
1039 0 : HcclResult ret = WaitCommandWithOpId(retryCtx->agentSocket_, commandInfo);
1040 0 : if (ret == HCCL_SUCCESS) {
1041 0 : if (commandInfo.command == RETRY_CMD_RESUME_TRANSPORT) {
1042 0 : HCCL_RUN_INFO("[OpRetry][Agent][Resume]WaitResumeCmdChangeLink, recv command[%s], group[%s], rankId[%u]",
1043 : GetReadableCmd(commandInfo.command), retryCtx->group_.c_str(), retryCtx->rankId_);
1044 0 : break;
1045 : }
1046 : }
1047 0 : }
1048 0 : return HCCL_SUCCESS;
1049 : }
1050 :
1051 0 : HcclResult ResumeAgentChangeLink::WaitAndRespLinkChanged(RetryContext *retryCtx, RetryState &nextState)
1052 : {
1053 0 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
1054 0 : const std::chrono::seconds timeout = std::chrono::seconds(OP_RETRY_WAIT_AICPU_TIMEOUT);
1055 0 : HCCL_INFO("[OpRetry][Agent][Resume]WaitAndRespLinkChanged start, rank[%d], group[%s]",
1056 : retryCtx->rankId_, retryCtx->group_.c_str());
1057 : while (true) {
1058 0 : KfcExecStatus &opInfo = retryCtx->localRetryInfo_.opInfo;
1059 0 : CHK_RET(GetOpExecInfo(retryCtx->GetD2hPtr(), opInfo));
1060 0 : const KfcStatus &aicpuState = opInfo.execStatus.kfcStatus;
1061 0 : if (aicpuState == KfcStatus::kResumeChanged) {
1062 0 : nextState = RETRY_STATE_AGENT_RUNNING;
1063 0 : CHK_RET(SetOpExecCmd(retryCtx->GetH2dPtr(), KfcCommand::kNone));
1064 0 : HCCL_INFO("[OpRetry][Agent][Resume]WaitAndRespLinkChanged, aicpuState[%d], rank[%d], group[%s]",
1065 : aicpuState, retryCtx->rankId_, retryCtx->group_.c_str());
1066 0 : break;
1067 : }
1068 : // 超时机制
1069 0 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
1070 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
1071 0 : CHK_PRT_BREAK(elapsed >= timeout,
1072 : HCCL_ERROR("[OpRetry][Agent][Resume]WaitAndRespLinkChanged timeout, aicpuState[%d]", aicpuState),
1073 : nextState = RETRY_STATE_RESP_RUNNING_ERR);
1074 : // 轮询间隔
1075 0 : SaluSleep(OP_RETRY_POLL_AICPU_STATE_INTERVAL);
1076 0 : }
1077 0 : if (nextState == RETRY_STATE_AGENT_RUNNING) {
1078 : // 回复成功
1079 0 : retryCtx->localRetryInfo_.retryState = RETRY_STATE_AGENT_RUNNING;
1080 0 : HCCL_INFO("[OpRetry][Agent][Resume]WaitAndRespLinkChanged success, rank[%d], group[%s]", retryCtx->rankId_, retryCtx->group_.c_str());
1081 0 : retryCtx->localRetryInfo_.opInfo.execStatus.kfcStatus = KfcStatus::kResumeChanged;
1082 0 : HcclResult ret = IssueResponse(retryCtx->agentSocket_, retryCtx->localRetryInfo_);
1083 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][Agent]ResumeAgentChangeLink IssueResponse fail"), ret);
1084 : }
1085 0 : return HCCL_SUCCESS;
1086 : }
1087 : } // namespace hccl
|
