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 "comm_configer.h"
13 : #include "externalinput_pub.h"
14 : #include "opretry_manager.h"
15 : #include "adapter_pub.h"
16 : #include "rt_external.h"
17 : #include "opretry_agent.h"
18 : #include "opretry_server.h"
19 : #include "opretry_base.h"
20 : #include "snapshot_control.h"
21 :
22 : #ifdef __cplusplus
23 : extern "C" {
24 : #endif // __cplusplus
25 :
26 : typedef enum tagRtClearStep {
27 : RT_STREAM_STOP = 0,
28 : RT_STREAM_CLEAR,
29 : } rtClearStep_t;
30 :
31 : extern rtError_t rtStreamClear(rtStream_t stm, rtClearStep_t step);
32 : #ifdef __cplusplus
33 : }
34 : #endif // __cplusplus
35 :
36 : namespace {
37 0 : HcclResult StreamClear(HcclRtStream stream, HcclRtStreamClearStep step)
38 : {
39 0 : CHK_PTR_NULL(stream);
40 :
41 : aclError ret;
42 0 : if (step == HcclRtStreamClearStep::HCCL_STREAM_STOP) {
43 0 : ret = rtStreamClear(stream, rtClearStep_t::RT_STREAM_STOP);
44 0 : HCCL_INFO("[StreamClear]Call rtStreamClear, ret[%d], param: stream[%p], step[%d]", ret, stream, step);
45 : } else {
46 0 : ret = rtStreamClear(stream, rtClearStep_t::RT_STREAM_CLEAR);
47 0 : HCCL_INFO("[StreamClear]Call rtStreamClear, ret[%d], param: stream[%p], step[%d]", ret, stream, step);
48 : }
49 0 : CHK_PRT_RET(ret != ACL_SUCCESS, HCCL_ERROR("[StreamClear]errNo[0x%016llx]Failed to clear stream. "
50 : "ret[%d], param: stream[%p], step[%d]", HCCL_ERROR_CODE(HCCL_E_RUNTIME), ret, stream, step), HCCL_E_RUNTIME);
51 0 : return HCCL_SUCCESS;
52 : }
53 : }
54 :
55 : namespace hccl {
56 0 : HcclResult OpRetryBase::Handle(RetryContext* retryCtx)
57 : {
58 0 : CheckSnapshotStatus(retryCtx);
59 0 : if (retryCtx->IsPaused()) {
60 0 : SaluSleep(OP_RETRY_RUNNING_POLL_INTERVAL);
61 0 : return HCCL_SUCCESS;
62 : }
63 :
64 0 : if (!retryCtx->IsRootRetryCtx() && retryCtx->isAgentStateWaitResume_ && retryCtx->GetRetryState() != RETRY_STATE_AGENT_WAIT_RESUME) {
65 0 : std::shared_ptr<OpRetryBase> retryPtr = nullptr;
66 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryAgentWaitResume>(), return HCCL_E_PTR);
67 0 : retryCtx->SetRetryState(RETRY_STATE_AGENT_WAIT_RESUME, retryPtr);
68 0 : retryCtx->ResetAgentState();
69 0 : HCCL_INFO("[OpRetry][Agent]switch to wait resume.");
70 0 : return HCCL_SUCCESS;
71 0 : }
72 :
73 0 : if (retryCtx->IsRootRetryCtx() && retryCtx->isServerStateWaitResume_ && retryCtx->GetRetryState() != RETRY_STATE_SERVER_WAIT_RESUME) {
74 0 : std::shared_ptr<OpRetryBase> retryPtr = nullptr;
75 0 : EXCEPTION_CATCH(retryPtr = std::make_shared<OpRetryServerWaitResume>(), return HCCL_E_PTR);
76 0 : retryCtx->SetRetryState(RETRY_STATE_SERVER_WAIT_RESUME, retryPtr);
77 0 : retryCtx->ResetServerState();
78 0 : HCCL_INFO("[OpRetry][Server]switch to wait resume.");
79 0 : return HCCL_SUCCESS;
80 0 : }
81 :
82 0 : HcclResult ret = ProcessEvent(retryCtx);
83 0 : if (ret != HCCL_SUCCESS) {
84 0 : CHK_RET(ProcessError(retryCtx));
85 0 : retryCtx->isOpRetryQuit = true;
86 : }
87 0 : return ret;
88 : }
89 :
90 : /* root-host 交互 */
91 0 : HcclResult OpRetryBase::IssueResponse(std::shared_ptr<HcclSocket> socket, RetryInfo &retryInfo)
92 : {
93 0 : return Send(socket, &retryInfo, sizeof(RetryInfo));
94 : }
95 :
96 : // 非阻塞接收, 若已经收到部分数据, 则变为阻塞接收, 直到收到完整数据或超时
97 1 : HcclResult OpRetryBase::WaitResponse(std::shared_ptr<HcclSocket> socket, RetryInfo &retryInfo)
98 : {
99 1 : HcclResult ret = Recv(socket, &retryInfo, sizeof(RetryInfo));
100 1 : if (ret == HCCL_SUCCESS) {
101 1 : HCCL_DEBUG("[OpRetry]WaitResponse success, cmd[%u]", retryInfo.cmd);
102 : }
103 1 : return ret;
104 : }
105 :
106 0 : HcclResult OpRetryBase::IssueCommand(std::shared_ptr<HcclSocket> socket, RetryCommand command)
107 : {
108 0 : HcclResult ret = Send(socket, &command, sizeof(RetryCommand));
109 0 : if (ret == HCCL_SUCCESS) {
110 0 : HCCL_DEBUG("[OpRetry]IssueCommand success, command[%s]", GetReadableCmd(command));
111 : }
112 0 : return ret;
113 : }
114 :
115 0 : HcclResult OpRetryBase::WaitCommand(std::shared_ptr<HcclSocket> socket, RetryCommand &command)
116 : {
117 0 : HcclResult ret = Recv(socket, &command, sizeof(RetryCommand));
118 0 : if (ret == HCCL_SUCCESS) {
119 0 : HCCL_DEBUG("[OpRetry]WaitCommand success, command[%s]", GetReadableCmd(command));
120 : }
121 0 : return ret;
122 : }
123 :
124 1 : HcclResult OpRetryBase::IssueCommandWithOpId(std::shared_ptr<HcclSocket> socket, RetryCommandInfo &commandInfo)
125 : {
126 1 : HcclResult ret = Send(socket, &commandInfo, sizeof(RetryCommandInfo));
127 1 : if (ret == HCCL_SUCCESS) {
128 1 : HCCL_DEBUG("[OpRetry]IssueCommand success, command[%s], cmd[%u]",
129 : GetReadableCmd(commandInfo.command), commandInfo.cmd);
130 : }
131 1 : return ret;
132 : }
133 :
134 1 : HcclResult OpRetryBase::WaitCommandWithOpId(std::shared_ptr<HcclSocket> socket, RetryCommandInfo &commandInfo)
135 : {
136 1 : HcclResult ret = Recv(socket, &commandInfo, sizeof(commandInfo));
137 1 : if (ret == HCCL_SUCCESS) {
138 1 : HCCL_DEBUG("[OpRetry]WaitCommand success, command[%s], cmd[%u]",
139 : GetReadableCmd(commandInfo.command), commandInfo.cmd);
140 : }
141 1 : return ret;
142 : }
143 :
144 0 : HcclResult OpRetryBase::IssueLinkPortCheckResult(std::shared_ptr<HcclSocket> socket, LinkPortStatus &linkPortStatus)
145 : {
146 0 : HcclResult ret = Send(socket, &linkPortStatus, sizeof(LinkPortStatus));
147 0 : if (ret == HCCL_SUCCESS) {
148 0 : HCCL_DEBUG("[OpRetry]IssueLinkPortCheckResult success");
149 : }
150 0 : return ret;
151 : }
152 :
153 0 : HcclResult OpRetryBase::WaitLinkPortCheckResult(std::shared_ptr<HcclSocket> socket, LinkPortStatus &linkPortStatus)
154 : {
155 0 : HcclResult ret = Recv(socket, &linkPortStatus, sizeof(LinkPortStatus));
156 0 : if (ret == HCCL_SUCCESS) {
157 0 : HCCL_DEBUG("[OpRetry]WaitLinkPortCheckResult success, cmd[%u]", linkPortStatus.cmd);
158 : }
159 0 : return ret;
160 : }
161 :
162 0 : HcclResult OpRetryBase::IssueChangeLink(std::shared_ptr<HcclSocket> socket, ChangeLinkInfo &changeLinkInfo)
163 : {
164 0 : HcclResult ret = Send(socket, &changeLinkInfo, sizeof(ChangeLinkInfo));
165 0 : if (ret == HCCL_SUCCESS) {
166 0 : HCCL_DEBUG("[OpRetry]IssueChangeLink success");
167 : }
168 0 : return ret;
169 : }
170 :
171 0 : HcclResult OpRetryBase::WaitChangeLink(std::shared_ptr<HcclSocket> socket, ChangeLinkInfo &changeLinkInfo)
172 : {
173 0 : HcclResult ret = Recv(socket, &changeLinkInfo, sizeof(ChangeLinkInfo));
174 0 : if (ret == HCCL_SUCCESS) {
175 0 : HCCL_DEBUG("[OpRetry]WaitChangeLink success");
176 : }
177 0 : return ret;
178 : }
179 :
180 : /* 校验 */
181 0 : HcclResult OpRetryBase::CheckRetryInfo(RetryContext &retryCtx)
182 : {
183 0 : for (auto rank : retryCtx.needRetryServerRanks_) {
184 : // 校验opName一致性
185 0 : auto &retryInfoStand = retryCtx.serverSockets_[*(retryCtx.needRetryServerRanks_.begin())].retryInfo;
186 0 : auto &retryInfo = retryCtx.serverSockets_[rank].retryInfo;
187 0 : u32 retryCnt = retryInfo.opInfo.execStatus.retryInfo.retryCount;
188 0 : HCCL_RUN_INFO("[OpRetry][Server][CheckRetryInfo]rankId[%u], opName[%s], index[%u], retryCnt[%u], linkState[%d]",
189 : retryInfo.rankId, retryInfo.opInfo.opId.tag, retryInfo.opInfo.opId.index,
190 : retryCnt, retryInfo.linkState);
191 :
192 0 : CHK_RET(CheckOpName(retryInfo, retryInfoStand));
193 : // 校验重传次数
194 0 : CHK_RET(CheckMaxRetryCnt(retryInfo, retryCtx.group_));
195 : // 校验链路状态
196 0 : CHK_RET(CheckLinkStates(retryInfo));
197 : }
198 0 : return HCCL_SUCCESS;
199 : }
200 :
201 0 : HcclResult OpRetryBase::CheckOpName(const RetryInfo &retryInfo1, const RetryInfo &retryInfo2)
202 : {
203 0 : if (retryInfo1.opInfo.opId.isSendRecv == true && retryInfo2.opInfo.opId.isSendRecv == true){
204 0 : const char* tag1 = reinterpret_cast<const char*>(retryInfo1.opInfo.opId.tag);
205 0 : const char* tag2 = reinterpret_cast<const char*>(retryInfo2.opInfo.opId.tag);
206 0 : const char* tag1Send = reinterpret_cast<const char*>(retryInfo1.opInfo.opId.bsrInfo[HCCL_SEND].bsrTag);
207 0 : const char* tag1Recv = reinterpret_cast<const char*>(retryInfo1.opInfo.opId.bsrInfo[HCCL_RECV].bsrTag);
208 0 : const char* tag2Send = reinterpret_cast<const char*>(retryInfo2.opInfo.opId.bsrInfo[HCCL_SEND].bsrTag);
209 0 : const char* tag2Recv = reinterpret_cast<const char*>(retryInfo2.opInfo.opId.bsrInfo[HCCL_RECV].bsrTag);
210 0 : u32 index1 = retryInfo1.opInfo.opId.index;
211 0 : u32 index2 = retryInfo2.opInfo.opId.index;
212 0 : u32 index1Send = retryInfo1.opInfo.opId.bsrInfo[HCCL_SEND].index;
213 0 : u32 index1Recv = retryInfo1.opInfo.opId.bsrInfo[HCCL_RECV].index;
214 0 : u32 index2Send = retryInfo2.opInfo.opId.bsrInfo[HCCL_SEND].index;
215 0 : u32 index2Recv = retryInfo2.opInfo.opId.bsrInfo[HCCL_RECV].index;
216 0 : bool isEqual = (((strcmp(tag1Send, tag2Recv) == 0) && (index1Send == index2Recv)) ||
217 0 : ((strcmp(tag1Recv, tag2Send) == 0) && (index1Recv == index2Send)) ||
218 0 : ((strcmp(tag1, tag2) == 0) && (index1 == index2)));
219 0 : CHK_PRT_RET(isEqual == false,
220 : HCCL_ERROR("[OpRetry][CheckOpName]hccl aicpu can not retry, opName is inconsistent: "\
221 : "rank[%u] tag1[%s] index1[%u] Stag1[%s] Sindex1[%u], Rtag1[%s] Rindex1[%u], IpInfo1[%s], "
222 : "rank[%u] tag2[%s] index2[%u] Stag2[%s] Sindex2[%u], Rtag2[%s] Rindex2[%u], IpInfo2[%s]",
223 : retryInfo1.rankId, tag1, index1, tag1Send, index1Send, tag1Recv, index1Recv, retryInfo1.dfxIpInfo,
224 : retryInfo2.rankId, tag1, index2, tag2Send, index2Send, tag2Recv, index2Recv, retryInfo2.dfxIpInfo),
225 : HCCL_E_OPRETRY_FAIL);
226 0 : return HCCL_SUCCESS;
227 : }
228 0 : const char* tag1 = reinterpret_cast<const char*>(retryInfo1.opInfo.opId.tag);
229 0 : const char* tag2 = reinterpret_cast<const char*>(retryInfo2.opInfo.opId.tag);
230 0 : u32 index1 = retryInfo1.opInfo.opId.index;
231 0 : u32 index2 = retryInfo2.opInfo.opId.index;
232 0 : bool isEqual = (strcmp(tag1, tag2) == 0) && (index1 == index2);
233 0 : CHK_PRT_RET(isEqual == false,
234 : HCCL_ERROR("[OpRetry][CheckOpName]hccl aicpu can not retry, opName is inconsistent: "\
235 : "rank[%u] tag1[%s] index1[%u] IpInfo1[%s], rank[%u] tag2[%s] index2[%u], IpInfo2[%s]",
236 : retryInfo1.rankId, tag1, index1, retryInfo1.dfxIpInfo, retryInfo2.rankId, tag2, index2,
237 : retryInfo2.dfxIpInfo), HCCL_E_OPRETRY_FAIL);
238 0 : return HCCL_SUCCESS;
239 : }
240 :
241 0 : HcclResult OpRetryBase::CheckMaxRetryCnt(const RetryInfo &retryInfo, const std::string& identifier)
242 : {
243 0 : u32 retryCount = retryInfo.opInfo.execStatus.retryInfo.retryCount;
244 0 : u32 retryMaxCnt = CommConfiger::GetInstance().GetCommConfigRetryMaxCnt(identifier);
245 0 : if (retryCount >= retryMaxCnt) {
246 0 : HCCL_ERROR("[OpRetry][CheckMaxRetryCnt]hccl aicpu can not retry, the retryCnt[%u] of rank[%u] with IpInfo[%s] "\
247 : "exceeds the MaxCnt[%u]", retryCount, retryInfo.rankId, retryInfo.dfxIpInfo, retryMaxCnt);
248 0 : return HCCL_E_PARA;
249 : }
250 0 : return HCCL_SUCCESS;
251 : }
252 :
253 0 : HcclResult OpRetryBase::CheckLinkStates(const RetryInfo &retryInfo)
254 : {
255 0 : if (retryInfo.linkState == false) {
256 0 : HCCL_ERROR("[OpRetry][CheckLinkStates]hccl aicpu can not retry, the linkState[%u] of rank[%u] with IpInfo[%s] "
257 : "should be %u", retryInfo.linkState, retryInfo.rankId, retryInfo.dfxIpInfo, true);
258 0 : return HCCL_E_PARA;
259 : }
260 0 : return HCCL_SUCCESS;
261 : }
262 :
263 : /* host-device 交互 */
264 0 : HcclResult OpRetryBase::GetRetryInfo(RetryContext* retryCtx, RetryInfo &retryInfo)
265 : {
266 0 : CHK_PTR_NULL(retryCtx);
267 :
268 0 : retryInfo.rankId = retryCtx->GetRankId();
269 0 : retryInfo.retryState = retryCtx->GetRetryState();
270 0 : retryInfo.linkState = true;
271 0 : CHK_RET(GetOpExecInfo(retryCtx->GetD2hPtr(), retryInfo.opInfo));
272 :
273 0 : HCCL_DEBUG("[OpRetry][GetRetryInfo]rankId[%u], retryState[%d], linkState[%d]",
274 : retryInfo.rankId, retryInfo.retryState, retryInfo.linkState);
275 :
276 0 : KfcExecStatus opInfo = retryInfo.opInfo;
277 0 : HCCL_DEBUG("[OpRetry][GetRetryInfo]tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d], opExeState[%d], "
278 : "errorCode[%d], retryCount[%u], streamid[%u]",
279 : opInfo.opId.tag, opInfo.opId.index, opInfo.opId.srcRank, opInfo.opId.detRank, opInfo.opId.isSendRecv,
280 : opInfo.execStatus.kfcStatus, opInfo.execStatus.kfcError, opInfo.execStatus.retryInfo.retryCount, opInfo.opId.streamId);
281 :
282 0 : return HCCL_SUCCESS;
283 : }
284 :
285 0 : HcclResult OpRetryBase::GetOpExecInfo(std::shared_ptr<HDCommunicate> hdcPtr, KfcExecStatus &opInfo)
286 : {
287 0 : CHK_SMART_PTR_NULL(hdcPtr);
288 0 : CHK_RET(hdcPtr->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
289 0 : HCCL_DEBUG("[OpRetry][GetOpExecInfo]tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d], opExeState[%d], "
290 : "errorCode[%d], retryCount[%u]",
291 : opInfo.opId.tag, opInfo.opId.index, opInfo.opId.srcRank, opInfo.opId.detRank, opInfo.opId.isSendRecv,
292 : opInfo.execStatus.kfcStatus, opInfo.execStatus.kfcError, opInfo.execStatus.retryInfo.retryCount);
293 0 : return HCCL_SUCCESS;
294 : }
295 :
296 0 : HcclResult OpRetryBase::SetOpExecCmd(std::shared_ptr<HDCommunicate> hdcPtr, KfcCommand opCmd)
297 : {
298 0 : HCCL_RUN_INFO("[OpRetry][SetOpExecCmd]set KfcCommand[%d]", opCmd);
299 0 : CHK_SMART_PTR_NULL(hdcPtr);
300 0 : CHK_RET(hdcPtr->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
301 0 : return HCCL_SUCCESS;
302 : }
303 :
304 0 : HcclResult OpRetryBase::SetOpExecCmdWithOpId(std::shared_ptr<HDCommunicate> hdcPtr, KfcCommand opCmd,
305 : HcclOpIdentifier &opId)
306 : {
307 0 : HCCL_RUN_INFO("[OpRetry][SetOpExecCmd]set KfcCommand[%d]", opCmd);
308 0 : CHK_SMART_PTR_NULL(hdcPtr);
309 :
310 0 : HCCL_RUN_INFO("[OpRetry][SetOpExecCmdWithOpId]tag[%s], index[%u], srcRank[%u], detRank[%u], isSendRecv[%d], "
311 : "streamid[%u]",
312 : opId.tag, opId.index, opId.srcRank, opId.detRank, opId.isSendRecv, opId.streamId);
313 : // 发送KfcCommand命令到hdc buffer中
314 0 : CHK_RET(hdcPtr->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
315 : // 计算targetOp偏移，发送HcclOpIdentifier数据到hdc buffer中
316 0 : u32 targetOpStart = sizeof(KfcCommand) + sizeof(BackgroundCommand) + sizeof(HcclComSuspendingFlag);
317 0 : CHK_RET(hdcPtr->Put(targetOpStart, sizeof(HcclOpIdentifier), reinterpret_cast<uint8_t *>(&opId)));
318 0 : return HCCL_SUCCESS;
319 : }
320 :
321 0 : HcclResult OpRetryBase::SetOpChangeLinkInfo(std::shared_ptr<HDCommunicate> hdcPtr, KfcCommand opCmd,
322 : ChangeLinkInfo &changeLinkInfo)
323 : {
324 0 : CHK_SMART_PTR_NULL(hdcPtr);
325 :
326 : // 发送KfcCommand命令到hdc buffer中
327 0 : CHK_RET(hdcPtr->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
328 : // 计算changeLinkInfo偏移，发送HcclOpIdentifier数据到hdc buffer中
329 0 : u32 changeLinkInfoStart = sizeof(KfcCommand) + sizeof(BackgroundCommand) + sizeof(HcclComSuspendingFlag) +
330 : sizeof(HcclOpIdentifier);
331 0 : CHK_RET(hdcPtr->Put(changeLinkInfoStart, sizeof(ChangeLinkInfo), reinterpret_cast<uint8_t *>(&changeLinkInfo)));
332 0 : return HCCL_SUCCESS;
333 : }
334 :
335 0 : HcclResult OpRetryBase::ClearStream(std::shared_ptr<HcclOpStreamRes> opStreamPtr_, HcclRtStreamClearStep clearStep)
336 : {
337 0 : HCCL_INFO("[OpRetry][ClearStream]start");
338 0 : CHK_SMART_PTR_NULL(opStreamPtr_);
339 0 : CHK_PRT_RET(opStreamPtr_->empty(), HCCL_ERROR("[OpRetry][ClearStream]fail, stream is empty"), HCCL_E_PARA);
340 0 : for (auto it = opStreamPtr_->begin(); it != opStreamPtr_->end(); it++) {
341 0 : const std::string &tag = it->first;
342 0 : std::vector<Stream> &streams = it->second;
343 0 : HCCL_RUN_INFO("[OpRetry][Agent]ClearStream clearStep:%u, tag:%u", clearStep, tag.c_str());
344 0 : for (auto &stream : streams) {
345 0 : HCCL_RUN_INFO("[OpRetry][Agent]ClearStream streamId:%u", stream.id());
346 0 : HcclResult ret = StreamClear(stream.ptr(), clearStep);
347 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][ClearStream]rtStream clear failed, stm:%p, "
348 : "step:%d, ret[%d].", stream.ptr(), clearStep, ret), HCCL_E_RUNTIME);
349 : }
350 : }
351 0 : return HCCL_SUCCESS;
352 : }
353 :
354 0 : HcclResult OpRetryBase::ClearStreamWithOpId(std::shared_ptr<HcclOpStreamRes> opStreamPtr_, HcclRtStreamClearStep clearStep,
355 : HcclOpIdentifier &opId, HcclOpIdentifier &curOpId)
356 : {
357 0 : HCCL_INFO("[OpRetry][ClearStream]start");
358 0 : CHK_SMART_PTR_NULL(opStreamPtr_);
359 0 : CHK_PRT_RET(opStreamPtr_->empty(), HCCL_ERROR("[OpRetry][ClearStream]fail, stream is empty"), HCCL_E_PARA);
360 :
361 0 : if (curOpId.isSendRecv && curOpId.streamId != ~0u){
362 0 : if (std::string(reinterpret_cast<const char*>(curOpId.tag)) != std::string(reinterpret_cast<const char*>(opId.tag)) ||
363 0 : curOpId.index != opId.index)
364 : {
365 0 : HCCL_ERROR("[OpRetry][Agent]ClearStream clearStep tag is inconsistent, curtag[%s],cmdtag[%s], curindex[%u], "
366 : "cmdindex[%u]", curOpId.tag, opId.tag, curOpId.index, opId.index);
367 : }
368 0 : for (auto it = opStreamPtr_->begin(); it != opStreamPtr_->end(); it++) {
369 0 : const std::string &tag = it->first;
370 0 : std::vector<Stream> &streams = it->second;
371 0 : HCCL_RUN_INFO("[OpRetry][Agent]ClearStream clearStep:%u, tag:%s", clearStep, tag.c_str());
372 0 : for (auto &stream : streams) {
373 0 : if (static_cast<u32>(stream.id()) == curOpId.streamId){
374 0 : HCCL_RUN_INFO("[OpRetry][Agent]ClearStream streamId:%u", curOpId.streamId);
375 0 : HcclResult ret = StreamClear(stream.ptr(), clearStep);
376 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][ClearStream]rtStream clear failed, stm:%p, "
377 : "step:%d, ret[%d].", stream.ptr(), clearStep, ret), HCCL_E_RUNTIME);
378 : }
379 : }
380 : }
381 0 : }
382 : else {
383 0 : for (auto it = opStreamPtr_->begin(); it != opStreamPtr_->end(); it++) {
384 0 : const std::string &tag = it->first;
385 0 : std::vector<Stream> &streams = it->second;
386 0 : HCCL_RUN_INFO("[OpRetry][Agent]ClearStream clearStep:%u, tag:%u", clearStep, tag.c_str());
387 0 : for (auto &stream : streams) {
388 0 : HCCL_RUN_INFO("[OpRetry][Agent]ClearStream streamId:%u", stream.id());
389 0 : HcclResult ret = StreamClear(stream.ptr(), clearStep);
390 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OpRetry][ClearStream]rtStream clear failed, stm:%p, "
391 : "step:%d, ret[%d].", stream.ptr(), clearStep, ret), HCCL_E_RUNTIME);
392 : }
393 : }
394 : }
395 :
396 0 : return HCCL_SUCCESS;
397 : }
398 :
399 0 : HcclResult OpRetryBase::ResetNotify(RetryContext* retryCtx)
400 : {
401 0 : CHK_PTR_NULL(retryCtx);
402 :
403 0 : auto remoteRank = retryCtx->localRetryInfo_.rankId == retryCtx->localRetryInfo_.opInfo.opId.detRank ?
404 : retryCtx->localRetryInfo_.opInfo.opId.srcRank : retryCtx->localRetryInfo_.opInfo.opId.detRank;
405 :
406 : // send/recv场景下只对对端的notify重置，其他场景下需要重置全部notify
407 0 : return retryCtx->notifyResetCallback_(retryCtx->localRetryInfo_.opInfo.opId.isSendRecv,
408 0 : static_cast<s64>(remoteRank));
409 : }
410 :
411 0 : HcclResult OpRetryBase::SetTransportStatusForStop(RetryContext* retryCtx)
412 : {
413 0 : CHK_PTR_NULL(retryCtx);
414 : // 用于表示当前rank与对端是否走借轨
415 0 : std::map<u32, bool> isChangeLinkMap;
416 0 : bool isChangeLinkFlag = false;
417 0 : if (retryCtx->isRecivedCmdToCheckLink) {
418 0 : return retryCtx->setTransportReseumeStatusCallback_(retryCtx->lastLinkPortStatus_, isChangeLinkMap, isChangeLinkFlag, true);
419 : }
420 : // stop阶段对当前正在使用的link执行，使用lastLinkPortStatus_表示当前正在使用的网口情况，默认为true，使用主网口
421 0 : return retryCtx->setTransportStatusCallback_(retryCtx->localRetryInfo_.opInfo.opId, true,
422 0 : retryCtx->lastLinkPortStatus_, isChangeLinkMap, isChangeLinkFlag);
423 0 : }
424 :
425 0 : HcclResult OpRetryBase::SetTransportStatusForResume(RetryContext* retryCtx)
426 : {
427 0 : CHK_PTR_NULL(retryCtx);
428 : // 用于表示当前rank与对端是否走借轨
429 0 : std::map<u32, bool> isChangeLinkMap;
430 0 : bool isChangeLinkFlag = false;
431 : // resume阶段
432 0 : std::map<u32, bool> remoteRankPortMap;
433 0 : for (u32 i = 0; i < retryCtx->localChangeLinkInfo_.remoteRankNum; i++) {
434 0 : auto remoteRank = retryCtx->localChangeLinkInfo_.remoteRankList[i];
435 0 : auto isRemoteUseDefaultPort = retryCtx->localChangeLinkInfo_.isUseDefaultPort[i];
436 : // remoteRankPortMap用于表示链路切换后的网口使用情况
437 0 : remoteRankPortMap.insert({remoteRank, isRemoteUseDefaultPort});
438 : // 和上一次链路使用情况对比判断当前是否为借轨场景，同时更新lastLinkPortStatus_
439 0 : if (retryCtx->lastLinkPortStatus_.find(remoteRank) == retryCtx->lastLinkPortStatus_.end()) {
440 : // 若remoteRank不在lastLinkPortStatus_中，默认上一次使用的是主链路
441 0 : isChangeLinkMap.insert({remoteRank, !isRemoteUseDefaultPort});
442 0 : isChangeLinkFlag = isRemoteUseDefaultPort ? isChangeLinkFlag : true;
443 0 : retryCtx->lastLinkPortStatus_.insert({remoteRank, isRemoteUseDefaultPort});
444 0 : } else if (isRemoteUseDefaultPort == retryCtx->lastLinkPortStatus_[remoteRank]) {
445 : // 若本次和上一次使用同一个port，则属于原地重执行场景
446 0 : isChangeLinkMap.insert({remoteRank, false});
447 : } else {
448 : // 若本次和上一次使用不同port，则属于借轨场景，并更新到lastLinkPortStatus_中
449 0 : isChangeLinkMap.insert({remoteRank, true});
450 0 : isChangeLinkFlag = true;
451 0 : retryCtx->lastLinkPortStatus_[remoteRank] = isRemoteUseDefaultPort;
452 : }
453 : }
454 :
455 0 : std::string isChangeLinkMapStr = "";
456 0 : for (auto changeIt: isChangeLinkMap) {
457 0 : isChangeLinkMapStr += (std::to_string(changeIt.first) + ":" + std::to_string(changeIt.second) + ";");
458 : }
459 0 : HCCL_RUN_INFO("[OpRetry][Agent]isChangeLinkFlag[%d]:[%s]", isChangeLinkFlag, isChangeLinkMapStr.c_str());
460 :
461 0 : retryCtx->localRetryInfo_.isChangeLinkFlag = isChangeLinkFlag; // 向server上报
462 0 : retryCtx->localChangeLinkInfo_.isChangeLinkFlag = isChangeLinkFlag; // 向aicpu下发
463 :
464 0 : if (retryCtx->isRecivedCmdToCheckLink) {
465 0 : retryCtx->isRecivedCmdToCheckLink = false;
466 0 : return retryCtx->setTransportReseumeStatusCallback_(remoteRankPortMap, isChangeLinkMap, isChangeLinkFlag, false);
467 : }
468 0 : return retryCtx->setTransportStatusCallback_(retryCtx->localRetryInfo_.opInfo.opId, false,
469 0 : remoteRankPortMap, isChangeLinkMap, isChangeLinkFlag);
470 0 : }
471 :
472 0 : HcclResult OpRetryBase::Send(std::shared_ptr<HcclSocket> socket, void *data, u64 size)
473 : {
474 0 : HCCL_DEBUG("[OpRetry][Send]start, para: data[%p], size[%llu Byte]", data, size);
475 0 : const auto start = std::chrono::steady_clock::now();
476 0 : const u32 timeoutValue = std::max(static_cast<u32>(GetExternalInputHcclLinkTimeOut()), OP_RETRY_SEND_RECV_TIMEOUT) + OP_RETRY_WAIT_AICPU_TIMEOUT;
477 0 : const std::chrono::seconds timeout = std::chrono::seconds(timeoutValue);
478 :
479 0 : u64 restSize = size; // 待发送数据长度
480 : while (true) {
481 0 : CHK_PRT_RET(!enableSendRecv, HCCL_DEBUG("[OpRetry][Send]Exit send."), HCCL_SUCCESS);
482 0 : u64 sendDis = size - restSize;
483 0 : void* dataPtr = reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(data) + sendDis);
484 : /* 获取当前时间，如果耗时超过timeout，则返回错误 */
485 : const auto elapsed =
486 0 : std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start);
487 0 : CHK_PRT_RET(elapsed > timeout,
488 : HCCL_WARNING("Send fail, Wait timeout for sockets send, dataPtr[%p], restSize[%llu Byte]", dataPtr, restSize),
489 : HCCL_E_TIMEOUT);
490 :
491 0 : u64 compSize = 0; // 本次发送数据长度
492 0 : HcclResult ret = socket->ISend(dataPtr, restSize, compSize);
493 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
494 : HCCL_WARNING("Send fail, dataPtr[%p], restSize[%llu Byte], compSize[%llu]", dataPtr, restSize, compSize), ret);
495 :
496 0 : if (restSize == compSize) { // 数据发送完成
497 0 : HCCL_DEBUG("OpRetryBase send end");
498 0 : return HCCL_SUCCESS;
499 0 : } else if (restSize < compSize) {
500 0 : HCCL_ERROR("Send fail, restSize[%llu Byte], compSize[%llu Byte]", restSize, compSize);
501 0 : return HCCL_E_TCP_TRANSFER;
502 : }
503 0 : restSize -= compSize;
504 0 : }
505 : return HCCL_SUCCESS;
506 : }
507 :
508 0 : HcclResult OpRetryBase::Recv(std::shared_ptr<HcclSocket> socket, void *data, u64 totalSize)
509 : {
510 0 : const auto start = std::chrono::steady_clock::now();
511 0 : const u32 timeoutValue = std::max(static_cast<u32>(GetExternalInputHcclLinkTimeOut()), OP_RETRY_SEND_RECV_TIMEOUT) + OP_RETRY_WAIT_AICPU_TIMEOUT;
512 0 : const std::chrono::seconds timeout = std::chrono::seconds(timeoutValue);
513 :
514 0 : u64 recvSize = 0;
515 : while (true) {
516 0 : CHK_PRT_RET(!enableSendRecv, HCCL_DEBUG("[OpRetry][Recv]Exit recv."), HCCL_SUCCESS);
517 : // 超时判断
518 : const auto elapsed =
519 0 : std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start);
520 0 : CHK_PRT_RET(elapsed > timeout,
521 : HCCL_ERROR("[OpRetry]Recv timeout, data[%p], recvSize[%llu Byte], totalSize[%llu Byte]", data, recvSize, totalSize),
522 : HCCL_E_TIMEOUT);
523 :
524 0 : u64 compSize = 0; // 本次接收到的长度
525 0 : u64 resetSize = totalSize - recvSize; // 待接收长度
526 0 : void* recvPtr = reinterpret_cast<void *>(reinterpret_cast<uintptr_t>(data) + recvSize);
527 0 : HcclResult ret = socket->IRecv(recvPtr, resetSize, compSize);
528 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, , ret);
529 :
530 0 : recvSize += compSize;
531 0 : if (recvSize == 0) { // 未收到数据
532 0 : return HCCL_E_AGAIN;
533 0 : } else if (recvSize < totalSize) { // 数据未接收完, 继续等待该对端
534 0 : HCCL_DEBUG("[OpRetry]Recv not complete, recvSize[%llu Byte], totalSize[%llu Byte]",
535 : recvSize, totalSize);
536 0 : SaluSleep(OP_RETRY_SEND_RECV_INTERVAL);
537 0 : continue;
538 : } else { // 数据接收完成
539 0 : return HCCL_SUCCESS;
540 : }
541 0 : }
542 : return HCCL_SUCCESS;
543 : }
544 :
545 0 : HcclResult OpRetryBase::InitChangeLinkInfo(RetryContext* retryCtx, bool incre, bool isGetGroupAllRemoteRank)
546 : {
547 0 : std::string newTag = std::string(reinterpret_cast<const char*>(retryCtx->localRetryInfo_.opInfo.opId.newTag));
548 0 : std::vector<u32> rankList;
549 0 : auto ret = OpRetryManager::GetLinkInfoByIdentifier(retryCtx->deviceLogicId_, retryCtx->group_, newTag,
550 : rankList, isGetGroupAllRemoteRank);
551 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
552 : HCCL_ERROR("[OpRetry][Agent][InitChangeLinkInfo] GetLinkInfoByIdentifier failed: deviceLogicId[%d], "
553 : "identify[%s], tag[%s], isGetGroupAllRemoteRank[%d]", retryCtx->deviceLogicId_,
554 : retryCtx->group_.c_str(), newTag.c_str(), isGetGroupAllRemoteRank), ret);
555 0 : if (retryCtx->localRetryInfo_.opInfo.opId.isSendRecv) {
556 : // send/recv场景下仅需校验对端
557 0 : auto remoteRank = retryCtx->localRetryInfo_.rankId == retryCtx->localRetryInfo_.opInfo.opId.detRank ?
558 : retryCtx->localRetryInfo_.opInfo.opId.srcRank : retryCtx->localRetryInfo_.opInfo.opId.detRank;
559 0 : bool isFind = std::count(rankList.begin(), rankList.end(), remoteRank) > 0;
560 0 : rankList.clear();
561 0 : if (isFind) {
562 0 : rankList.push_back(remoteRank);
563 : }
564 : }
565 :
566 0 : if (incre) {
567 : // 增量场景
568 0 : for (u32 remoteRank: rankList) {
569 : // 若对端不在已有的链路切换列表中，则加入，且默认为true
570 0 : if (retryCtx->lastLinkPortStatus_.find(remoteRank) == retryCtx->lastLinkPortStatus_.end()) {
571 0 : u32 index = retryCtx->localChangeLinkInfo_.remoteRankNum;
572 0 : retryCtx->localChangeLinkInfo_.remoteRankNum++;
573 0 : retryCtx->localChangeLinkInfo_.remoteRankList[index] = remoteRank;
574 0 : retryCtx->localChangeLinkInfo_.isUseDefaultPort[index] = true;
575 :
576 0 : retryCtx->lastLinkPortStatus_.insert({remoteRank, true});
577 :
578 0 : HCCL_RUN_INFO("[OpRetry][Agnet]init changeLinkInfoStr add remoteRank[%u]", remoteRank);
579 : }
580 : }
581 : } else {
582 : // 首次初始化场景
583 0 : retryCtx->localChangeLinkInfo_.remoteRankNum = rankList.size();
584 0 : std::copy(rankList.begin(), rankList.end(), retryCtx->localChangeLinkInfo_.remoteRankList);
585 0 : CHK_SAFETY_FUNC_RET(memset_s(retryCtx->localChangeLinkInfo_.isUseDefaultPort, rankList.size(), true, rankList.size()));
586 :
587 : // agent的初始化changeLinkInfo信息
588 0 : std::string changeLinkInfoStr = "";
589 0 : for (u32 i = 0; i < retryCtx->localChangeLinkInfo_.remoteRankNum; i++) {
590 0 : changeLinkInfoStr += (std::to_string(retryCtx->localChangeLinkInfo_.remoteRankList[i]) + ":" +
591 0 : std::to_string(retryCtx->localChangeLinkInfo_.isUseDefaultPort[i]) + "; ");
592 0 : retryCtx->lastLinkPortStatus_.insert({retryCtx->localChangeLinkInfo_.remoteRankList[i], true});
593 : }
594 0 : HCCL_RUN_INFO("[OpRetry][Agnet]init changeLinkInfoStr:%s", changeLinkInfoStr.c_str());
595 0 : }
596 :
597 0 : return HCCL_SUCCESS;
598 0 : }
599 :
600 0 : HcclResult OpRetryBase::GetLinkPortStatus(RetryContext* retryCtx, LinkPortStatus &linkPortStatus,
601 : bool isGetGroupAllRemoteRank)
602 : {
603 0 : std::string newTag = std::string(reinterpret_cast<const char*>(retryCtx->localRetryInfo_.opInfo.opId.newTag));
604 0 : HCCL_RUN_INFO("[OpRetry][Agent]begin to GetLinkPortStatus from: deviceLogicId[%d], identifier[%s] tag[%s]",
605 : retryCtx->deviceLogicId_, retryCtx->group_.c_str(), newTag.c_str());
606 :
607 0 : std::vector<u32> rankList;
608 0 : auto ret = OpRetryManager::GetLinkInfoByIdentifier(retryCtx->deviceLogicId_, retryCtx->group_, newTag,
609 : rankList, isGetGroupAllRemoteRank);
610 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
611 : HCCL_ERROR("[OpRetry][Agent][GetLinkPortStatus] GetLinkInfoByIdentifier failed: deviceLogicId[%d], "
612 : "identify[%s], tag[%s], isGetGroupAllRemoteRank[%d]", retryCtx->deviceLogicId_, retryCtx->group_.c_str(),
613 : newTag.c_str(), isGetGroupAllRemoteRank), ret);
614 0 : if (retryCtx->localRetryInfo_.opInfo.opId.isSendRecv) {
615 : // send/recv场景下仅需校验对端
616 0 : auto remoteRank = retryCtx->localRetryInfo_.rankId == retryCtx->localRetryInfo_.opInfo.opId.detRank ?
617 : retryCtx->localRetryInfo_.opInfo.opId.srcRank : retryCtx->localRetryInfo_.opInfo.opId.detRank;
618 0 : bool isFind = std::count(rankList.begin(), rankList.end(), remoteRank) > 0;
619 0 : rankList.clear();
620 0 : if (isFind) {
621 0 : rankList.push_back(remoteRank);
622 : }
623 : }
624 0 : std::copy(rankList.begin(), rankList.end(), linkPortStatus.rankList);
625 0 : linkPortStatus.rankSize = rankList.size();
626 :
627 0 : if (rankList.size() == 0) {
628 : // 若rankList为空，则说明当前无roce连接，无需获取主备网口link状态
629 0 : HCCL_RUN_INFO("[OpRetry][Agent]deviceLogicId[%d], rankSize[%d], not need to get link port, identifier[%s], tag[%s]",
630 : retryCtx->deviceLogicId_, linkPortStatus.rankSize, retryCtx->group_.c_str(), newTag.c_str());
631 0 : return HCCL_SUCCESS;
632 : }
633 :
634 0 : if (retryCtx->isUseDefaultPort_) {
635 0 : CHK_RET(HcclNetDevGetPortStatus(retryCtx->netDevCtx_, linkPortStatus.defaultPort));
636 0 : if (!linkPortStatus.defaultPort) {
637 0 : HCCL_RUN_INFO("[OpRetry][Agent]defaultPort is down, set isUseDefaultPort_ to false");
638 0 : retryCtx->isUseDefaultPort_ = false;
639 : }
640 : } else {
641 : // 发生借轨后不检测主网口，暂不支持回切
642 0 : linkPortStatus.defaultPort = false;
643 0 : HCCL_RUN_INFO("[OpRetry][Agent]defaultPort is not enable after last check, set defaultPort to false");
644 : }
645 :
646 0 : if (retryCtx->isEnableBackupLink_) {
647 : // 使能借轨场景下才需要获取backupPort状态
648 0 : CHK_RET(HcclNetDevGetPortStatus(retryCtx->backUpNetDevCtx_, linkPortStatus.backupPort));
649 : } else {
650 : // 默认场景下backupPort状态为false
651 0 : linkPortStatus.backupPort = false;
652 0 : HCCL_RUN_INFO("[OpRetry][Agent]backUpLink is not enable, set backupPort to false");
653 : }
654 :
655 0 : HCCL_RUN_INFO("[OpRetry][Agent]GetLinkPortStatus success: deviceLogicId[%d], rankSize[%d], identifier[%s], tag[%s]",
656 : retryCtx->deviceLogicId_, linkPortStatus.rankSize, retryCtx->group_.c_str(), newTag.c_str());
657 0 : return HCCL_SUCCESS;
658 0 : }
659 0 : HcclResult OpRetryBase::SetBsrOpId(RetryContext* retryCtx, HcclSendRecvType type)
660 : {
661 0 : auto &opId = retryCtx->localRetryInfo_.opInfo.opId;
662 0 : auto &bsrOpId = (HcclSendRecvType::HCCL_SEND == type) ? retryCtx->RemainSendOpId_ : retryCtx->RemainRecvOpId_;
663 :
664 : //重执行需要的信息
665 0 : bsrOpId.index = opId.bsrInfo[type].index;
666 0 : CHK_SAFETY_FUNC_RET(memset_s(bsrOpId.tag, sizeof(bsrOpId.tag), 0, sizeof(bsrOpId.tag)));
667 0 : CHK_SAFETY_FUNC_RET(memcpy_s(bsrOpId.tag, sizeof(bsrOpId.tag), opId.bsrInfo[type].bsrTag,
668 : sizeof(opId.bsrInfo[type].bsrTag)));
669 0 : bsrOpId.srcRank = opId.bsrInfo[type].srcRank;
670 0 : bsrOpId.detRank = opId.bsrInfo[type].detRank;
671 0 : bsrOpId.streamId = opId.bsrInfo[type].streamId;
672 0 : bsrOpId.bsrInfo[type].tpQpn = opId.bsrInfo[type].tpQpn;
673 0 : bsrOpId.isSendRecv = true;
674 0 : bsrOpId.opType = HcclCMDType::HCCL_CMD_BATCH_SEND_RECV;
675 0 : return HCCL_SUCCESS;
676 : }
677 :
678 0 : HcclResult OpRetryBase::GetBsrOpId(RetryContext* retryCtx, HcclSendRecvType type)
679 : {
680 0 : auto &opId = retryCtx->localRetryInfo_.opInfo.opId;
681 0 : auto &bsrOpId = (HcclSendRecvType::HCCL_SEND == type) ? retryCtx->RemainSendOpId_ : retryCtx->RemainRecvOpId_;
682 : //重执行需要的信息
683 0 : opId.index = bsrOpId.index;
684 0 : CHK_SAFETY_FUNC_RET(memset_s(opId.tag, sizeof(opId.tag), 0, sizeof(opId.tag)));
685 0 : CHK_SAFETY_FUNC_RET(memcpy_s(opId.tag, sizeof(opId.tag), bsrOpId.tag, sizeof(bsrOpId.tag)));
686 0 : opId.srcRank = bsrOpId.srcRank;
687 0 : opId.detRank = bsrOpId.detRank;
688 0 : opId.streamId = bsrOpId.streamId;
689 0 : opId.isSendRecv = true;
690 0 : opId.opType = HcclCMDType::HCCL_CMD_BATCH_SEND_RECV;
691 0 : return HCCL_SUCCESS;
692 : }
693 :
694 1 : HcclResult OpRetryBase::IssueActiveSwitchInfo(std::shared_ptr<HcclSocket> socket, ActiveSwitchInfo &switchInfo)
695 : {
696 1 : HcclResult ret = Send(socket, &switchInfo, sizeof(ActiveSwitchInfo));
697 1 : if (ret == HCCL_SUCCESS) {
698 1 : HCCL_INFO("[SwitchNic][send] send success fin[%u], switchRankNm[%u], remoteRankNum[%u]",
699 : switchInfo.refreshTransportFin, switchInfo.switchRankNum, switchInfo.remoteRankNum);
700 : } else {
701 0 : HCCL_ERROR("[SwitchNic] send active switch info fail.");
702 : }
703 1 : return ret;
704 : }
705 :
706 3 : HcclResult OpRetryBase::WaitActiveSwitchInfo(std::shared_ptr<HcclSocket> socket, ActiveSwitchInfo &switchInfo)
707 : {
708 3 : HcclResult ret = Recv(socket, &switchInfo, sizeof(ActiveSwitchInfo));
709 3 : if (ret == HCCL_SUCCESS) {
710 2 : HCCL_INFO("[SwitchNic] recv success fin[%u], switchRankNm[%u], remoteRankNum[%u], cmd[%u]",
711 : switchInfo.refreshTransportFin, switchInfo.switchRankNum, switchInfo.remoteRankNum, switchInfo.cmd);
712 : }
713 3 : return ret;
714 : }
715 :
716 4 : HcclResult OpRetryBase::RecvActiveSwitchInfo(std::shared_ptr<HcclSocket> socket,
717 : const u32 rankId, ActiveSwitchInfo &switchInfo)
718 : {
719 4 : std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();
720 4 : const auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut() * ACTIVE_SWITCH_TIMES);
721 4 : HcclResult ret = HCCL_SUCCESS;
722 : while (true) {
723 : // 判断是否超时
724 5 : std::chrono::steady_clock::time_point curTime = std::chrono::steady_clock::now();
725 5 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(curTime - startTime);
726 5 : CHK_PRT_RET(elapsed > timeout,
727 : HCCL_ERROR("[SwitchNic][Server] timeout in recv agent ActiveSwitchInfo, waitime[%u>%u]",
728 : elapsed, timeout), HCCL_E_TIMEOUT);
729 5 : ret = WaitActiveSwitchInfo(socket, switchInfo);
730 5 : if (ret == HCCL_SUCCESS) {
731 4 : HCCL_INFO("[SwitchNic][Server] recv ActiveSwitchInfo form rank[%u]", rankId);
732 4 : break;
733 1 : } else if (ret == HCCL_E_AGAIN) {
734 : // 未收到数据，发送一个保活数据给agent
735 1 : RetryCommand command = RETRY_CMD_RUNNING;
736 1 : CHK_RET(IssueCommand(socket, command));
737 1 : HCCL_DEBUG("[SwitchNic][Server] send keeping active info to rank[%u]", rankId);
738 : } else {
739 0 : HCCL_ERROR("[SwitchNic][Server] failed to recv ActiveSwitchInfo, rank[%u], ret[%u]", rankId, ret);
740 0 : break;
741 : }
742 1 : SaluSleep(OP_RETRY_POLL_AICPU_STATE_INTERVAL);
743 1 : }
744 4 : return ret;
745 : }
746 :
747 0 : HcclResult OpRetryBase::GetSwitchRanks(RetryContext* retryCtx, bool &needCheckDefaultNic, bool &needCheckBackupNic)
748 : {
749 0 : CHK_PTR_NULL(retryCtx);
750 0 : return retryCtx->getSwitchRanksCallback_(retryCtx->switchInfo_.switchRankList,
751 0 : retryCtx->switchInfo_.switchUseBackup, retryCtx->switchInfo_.switchRankNum,
752 0 : retryCtx->switchInfo_.remoteRankNicStatus, retryCtx->switchInfo_.remoteRankNum,
753 0 : needCheckDefaultNic, needCheckBackupNic);
754 : }
755 :
756 0 : void OpRetryBase::SetEnableSendRecv(bool enable){
757 0 : enableSendRecv = enable;
758 0 : }
759 :
760 0 : void OpRetryBase::CheckSnapshotStatus(RetryContext* retryCtx) {
761 0 : auto snapshotStatus = SnapshotControl::GetInstance(retryCtx->deviceLogicId_).GetStatus();
762 0 : if (retryCtx->isPaused_ && snapshotStatus == SnapshotStatus::POST_SNAPSHOT) {
763 0 : retryCtx->isPaused_ = false;
764 0 : HCCL_RUN_INFO("[OpRetryBase][CheckSnapshotStatus] detect snapshot post-processing, "
765 : "opretry is resumed, curState[%s], deviceLogicId[%d].",
766 : retryCtx->GetReadableCtxState(), retryCtx->deviceLogicId_);
767 0 : } else if (!retryCtx->isPaused_ && snapshotStatus == SnapshotStatus::PRE_SNAPSHOT) {
768 0 : retryCtx->isPaused_ = true;
769 0 : HCCL_RUN_INFO("[OpRetryBase][CheckSnapshotStatus] detect snapshot pre-processing, "
770 : "opretry is paused, curState[%s], deviceLogicId[%d].",
771 : retryCtx->GetReadableCtxState(), retryCtx->deviceLogicId_);
772 : }
773 0 : }
774 : }
|
