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