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