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 : #ifndef HCCL_RETRY_BASE_H
12 : #define HCCL_RETRY_BASE_H
13 :
14 : #include <memory>
15 : #include "hccl_socket.h"
16 : #include "notify_pool.h"
17 : #include "hccl_op_retry_pub.h"
18 : #include "hdc_pub.h"
19 : #include "exception_handler.h"
20 : #include "hccl_common.h"
21 :
22 : namespace hccl {
23 : constexpr u32 OP_RETRY_MAX_CNT = 3;
24 : constexpr u32 OP_RETRY_WAIT_AICPU_TIMEOUT = 5; // 等待Aicpu的时长, 单位s
25 : constexpr u32 OP_RETRY_WAIT_AGENT_AICPU_TIMEOUT = 10; // 等待Agent+Aicpu的时长, 单位s
26 : constexpr u32 OP_RETRY_POLL_AICPU_ERROR_INTERVAL = 1; // 正常状态轮询Aicpu错误码的间隔, 单位s
27 : constexpr u32 OP_RETRY_POLL_RDMA_ERROR_INTERVAL = 1; // 正常状态轮询RDMA错误码的间隔, 单位s
28 : constexpr u32 OP_RETRY_POLL_AICPU_STATE_INTERVAL = 10000; // 重执行状态轮询Aicpu状态的间隔, 单位us
29 : constexpr u32 OP_RETRY_SEND_RECV_TIMEOUT = 200; // 发送和接收的超时时间, 单位s
30 : constexpr u32 OP_RETRY_SEND_RECV_INTERVAL = 10000; // 发送和接收的间隔时间, 单位us
31 : constexpr u32 OP_RETRY_KEEP_INTERVAL = 1; // 保活时间间隔, 单位s
32 : constexpr u32 OP_RETRY_RUNNING_POLL_INTERVAL = 100000; // 重执行状态轮询状态的间隔, 单位us
33 : constexpr u32 TIME_MS_TO_US = 1000;
34 :
35 : // 重执行初始化需要用到的参数
36 : struct OpRetryAgentParam {
37 : std::string group;
38 : std::shared_ptr<HcclSocket> agentConnection;
39 : std::shared_ptr<HDCommunicate> h2dPtr;
40 : std::shared_ptr<HDCommunicate> d2hPtr;
41 : std::shared_ptr<HcclOpStreamRes> opStreamPtr;
42 : OpRetryResetNotifyCallback notifyResetCallback;
43 : OpRetrySetTransportStatusCallback setTransportStatusCallback;
44 : OpRetryGetSwitchRanksCallback getSwitchRanksCallback;
45 : OpRetrySetTransportResumeStatusCallBack setTransportResumeStatusCallback;
46 : bool isEnableBackupLink;
47 : bool isEnableSdmaRetry;
48 : OpRetryAgentInfo agentInfo;
49 : };
50 :
51 3 : struct LinkPortStatus {
52 : u32 cmd = LINK_PORT_STATUS_CMD;
53 : bool defaultPort = false;
54 : bool backupPort = false;
55 : u32 rankSize = 0;
56 : u32 rankList[AICPU_MAX_RANK_NUM] = {};
57 : };
58 :
59 3 : struct ActiveSwitchInfo {
60 : u32 cmd = ACTIVE_SWITCH_INFO_CMD;
61 : u32 switchRankNum;
62 : u32 remoteRankNum;
63 : bool refreshTransportFin = false;
64 : bool defaultPortStatus = false;
65 : bool backupPortStatus = false;
66 : bool localPortsCheckRet = false;
67 : u32 switchRankList[AICPU_MAX_RANK_NUM] = {};
68 : bool switchUseBackup[AICPU_MAX_RANK_NUM] = {};
69 : u8 remoteRankNicStatus[AICPU_MAX_RANK_NUM] = {};
70 : };
71 :
72 5 : using HcclAgentRetryInfo = struct HcclAgentRetryInfoDef {
73 : std::shared_ptr<HcclSocket> socket{nullptr};
74 : RetryInfo retryInfo;
75 : ChangeLinkInfo changeLinkInfo;
76 : LinkPortStatus linkPortStatus;
77 : ActiveSwitchInfo switchInfo;
78 : };
79 :
80 57 : inline const char *GetReadableState(RetryState retryState) {
81 57 : auto it = RETRY_STATE_STR_MAP.find(retryState);
82 57 : return (it != RETRY_STATE_STR_MAP.end()) ? it->second.c_str() : "unknown state";
83 : }
84 :
85 4 : inline const char *GetReadableCmd(RetryCommand retryCommand) {
86 4 : auto it = RETRY_COMMAND_STR_MAP.find(retryCommand);
87 4 : return (it != RETRY_COMMAND_STR_MAP.end()) ? it->second.c_str() : "unknown cmd";
88 : }
89 :
90 : class RetryContext;
91 :
92 : // 状态基类
93 : class OpRetryBase {
94 : public:
95 : virtual HcclResult Handle(RetryContext* retryCtx);
96 : virtual HcclResult ProcessEvent(RetryContext* retryCtx) = 0;
97 : virtual HcclResult ProcessError(RetryContext* retryCtx) = 0;
98 :
99 36 : OpRetryBase() {};
100 36 : virtual ~OpRetryBase() {};
101 :
102 : // 设置是否直接退出send/recv循环状态，规避长时间阻塞，无法切换状态
103 : void SetEnableSendRecv(bool enable);
104 : protected:
105 : /* server-agent 交互 */
106 : HcclResult IssueResponse(std::shared_ptr<HcclSocket> socket, RetryInfo &retryInfo); // agent向server发送数据
107 : HcclResult WaitResponse(std::shared_ptr<HcclSocket> socket, RetryInfo &retryInfo); // server等待agent回复
108 :
109 : HcclResult IssueCommand(std::shared_ptr<HcclSocket> socket, RetryCommand command); // server向agent发送命令
110 : HcclResult WaitCommand(std::shared_ptr<HcclSocket> socket, RetryCommand &command); // agent轮询命令
111 :
112 : // server向agent发送命令,携带opid
113 : HcclResult IssueCommandWithOpId(std::shared_ptr<HcclSocket> socket, RetryCommandInfo &commandInfo);
114 : // agent轮询命令,携带opid
115 : HcclResult WaitCommandWithOpId(std::shared_ptr<HcclSocket> socket, RetryCommandInfo &commandInfo);
116 :
117 : // server向agent发送借轨信息
118 : HcclResult IssueChangeLink(std::shared_ptr<HcclSocket> socket, ChangeLinkInfo &changeLinkInfo);
119 : // agent轮询借轨信息
120 : HcclResult WaitChangeLink(std::shared_ptr<HcclSocket> socket, ChangeLinkInfo &changeLinkInfo);
121 : // agent向server发送当前网口情况
122 : HcclResult IssueLinkPortCheckResult(std::shared_ptr<HcclSocket> socket, LinkPortStatus &linkPortStatus);
123 : // server接收当前网口情况
124 : HcclResult WaitLinkPortCheckResult(std::shared_ptr<HcclSocket> socket, LinkPortStatus &linkPortStatus);
125 : // agent向device发送借轨信息
126 : HcclResult SetOpChangeLinkInfo(std::shared_ptr<HDCommunicate> hdcPtr, KfcCommand opCmd,
127 : ChangeLinkInfo &changeLinkInfo);
128 : // agent向server发送主动借轨信息
129 : HcclResult IssueActiveSwitchInfo(std::shared_ptr<HcclSocket> socket, ActiveSwitchInfo &switchInfo);
130 : // server收到主动借轨信息
131 : HcclResult WaitActiveSwitchInfo(std::shared_ptr<HcclSocket> socket, ActiveSwitchInfo &switchInfo);
132 : // server处理收到主动借轨信息函数
133 : HcclResult RecvActiveSwitchInfo(std::shared_ptr<HcclSocket> socket, const u32 rankId, ActiveSwitchInfo &switchInfo);
134 :
135 : // 获取SwitchRanks等信息
136 : HcclResult GetSwitchRanks(RetryContext* retryCtx, bool &needCheckDefaultNic, bool &needCheckBackupNic);
137 :
138 : /* 校验 */
139 : HcclResult CheckRetryInfo(RetryContext &retryCtx); // 校验收到的N个RetryInfo
140 : HcclResult GetRetryInfo(RetryContext* retryCtx, RetryInfo &retryInfo);
141 :
142 : /* agent-device 交互 */
143 : HcclResult GetOpExecInfo(std::shared_ptr<HDCommunicate> hdcPtr, KfcExecStatus &opInfo);
144 : HcclResult SetOpExecCmd(std::shared_ptr<HDCommunicate> hdcPtr, KfcCommand opCmd);
145 : HcclResult ClearStream(std::shared_ptr<HcclOpStreamRes> opStreamPtr_, HcclRtStreamClearStep clearStep);
146 : HcclResult SetOpExecCmdWithOpId(std::shared_ptr<HDCommunicate> hdcPtr, KfcCommand opCmd, HcclOpIdentifier &opId);
147 : HcclResult ClearStreamWithOpId(std::shared_ptr<HcclOpStreamRes> opStreamPtr_, HcclRtStreamClearStep clearStep,
148 : HcclOpIdentifier &opId, HcclOpIdentifier &curOpId);
149 : HcclResult ResetNotify(RetryContext* retryCtx);
150 : HcclResult SetTransportStatusForStop(RetryContext* retryCtx);
151 : HcclResult SetTransportStatusForResume(RetryContext* retryCtx);
152 : HcclResult GetLinkPortStatus(RetryContext* retryCtx, LinkPortStatus &linkPortStatus,
153 : bool isGetGroupAllRemoteRank = false);
154 : HcclResult InitChangeLinkInfo(RetryContext* retryCtx, bool incre = false, bool isGetGroupAllRemoteRank = false);
155 : /*获取batchsendrecv rdma重执行时的故障信息*/
156 : HcclResult SetBsrOpId(RetryContext* retryCtx, HcclSendRecvType type);
157 : HcclResult GetBsrOpId(RetryContext* retryCtx, HcclSendRecvType type);
158 : private:
159 : // 阻塞式发送 && 非阻塞式接收, 接口内部不报错, 返回值在上层判断并打印日志, 避免未进入重执行时出现ERROR日志
160 : HcclResult Send(std::shared_ptr<HcclSocket> socket, void *data, u64 size);
161 : HcclResult Recv(std::shared_ptr<HcclSocket> socket, void *data, u64 size);
162 :
163 : HcclResult CheckOpName(const RetryInfo &opInfo1, const RetryInfo &opInfo2); // 校验算子一致
164 : HcclResult CheckMaxRetryCnt(const RetryInfo &retryInfo, const std::string& identifier = HCCL_WORLD_GROUP); // 校验重执行次数
165 : HcclResult CheckLinkStates(const RetryInfo &retryInfo); // 校验link状态
166 : void CheckSnapshotStatus(RetryContext* retryCtx);
167 : bool enableSendRecv = true;
168 : };
169 :
170 : class RetryContext {
171 : public:
172 : // agent状态机初始化
173 6 : RetryContext(OpRetryAgentParam ¶m, std::shared_ptr<OpRetryBase> retryBase)
174 18 : {
175 6 : group_ = param.group;
176 6 : agentSocket_ = param.agentConnection;
177 6 : h2dPtr_ = param.h2dPtr;
178 6 : d2hPtr_ = param.d2hPtr;
179 6 : opStreamPtr_ = param.opStreamPtr;
180 6 : notifyResetCallback_ = param.notifyResetCallback;
181 6 : setTransportStatusCallback_ = param.setTransportStatusCallback;
182 6 : getSwitchRanksCallback_ = param.getSwitchRanksCallback;
183 6 : setTransportReseumeStatusCallback_ = param.setTransportResumeStatusCallback;
184 6 : isEnableBackupLink_ = param.isEnableBackupLink;
185 6 : isEnableSdmaRetry_ = param.isEnableSdmaRetry;
186 6 : retryBase_ = retryBase;
187 6 : isRootRetryCtx_ = false;
188 :
189 6 : rankId_ = param.agentInfo.userRank;
190 6 : deviceLogicId_ = param.agentInfo.deviceLogicId;
191 6 : netDevCtx_ = param.agentInfo.netDevCtx;
192 6 : backUpNetDevCtx_ = param.agentInfo.backUpNetDevCtx;
193 12 : std::string dfxInfo = "deviceIP:" + std::string(param.agentInfo.deviceIP.GetReadableIP()) +
194 18 : ";hostIP:" + std::string(param.agentInfo.hostIP.GetReadableIP());
195 6 : EXCEPTION_THROW_IF_COND_ERR(memcpy_s(localRetryInfo_.dfxIpInfo, sizeof(localRetryInfo_.dfxIpInfo),
196 : dfxInfo.c_str(), dfxInfo.size()) != EOK, "memcpy_s dfxIpInfo failed.");
197 6 : localRetryInfo_.dfxIpInfo[dfxInfo.size()] = '\0';
198 6 : }
199 :
200 : // server状态机初始化
201 8 : RetryContext(std::map<u32, std::shared_ptr<HcclSocket> > &sockets,
202 : std::shared_ptr<OpRetryBase> retryBase, const OpRetryAgentInfo& agentInfo)
203 24 : {
204 8 : retryBase_ = retryBase;
205 8 : isRootRetryCtx_ = true;
206 20 : for (auto it = sockets.begin(); it != sockets.end(); ++it) {
207 12 : HcclAgentRetryInfo tempAgentInfo;
208 12 : tempAgentInfo.socket = it->second;
209 14 : serverSockets_.insert(std::make_pair(it->first, std::move(tempAgentInfo)));
210 10 : }
211 8 : rankId_ = agentInfo.userRank;
212 8 : deviceLogicId_ = agentInfo.deviceLogicId;
213 16 : std::string dfxInfo = "deviceIP:" + std::string(agentInfo.deviceIP.GetReadableIP()) +
214 21 : ",hostIP:" + std::string(agentInfo.hostIP.GetReadableIP());
215 8 : EXCEPTION_THROW_IF_COND_ERR(memcpy_s(localRetryInfo_.dfxIpInfo, sizeof(localRetryInfo_.dfxIpInfo),
216 : dfxInfo.c_str(), dfxInfo.size()) != EOK, "memcpy_s dfxIpInfo failed.");
217 8 : localRetryInfo_.dfxIpInfo[dfxInfo.size()] = '\0';
218 8 : }
219 :
220 1 : RetryState GetRetryState() {
221 1 : return state_;
222 : }
223 0 : const char *GetReadableCtxState() const {
224 0 : return GetReadableState(state_);
225 : }
226 :
227 18 : void SetRetryState(RetryState nextState, std::shared_ptr<OpRetryBase> retryBase) {
228 18 : HCCL_RUN_INFO("[OpRetry][%s]State Transfer, cur state %s, next state %s",
229 : GetOpRetryMachineType(), GetReadableState(state_), GetReadableState(nextState));
230 18 : state_ = nextState;
231 18 : retryBase_ = retryBase;
232 18 : localRetryInfo_.retryState = state_;
233 18 : }
234 :
235 : void SetEnableSendRecv(bool enable) {
236 : retryBase_->SetEnableSendRecv(enable);
237 : };
238 :
239 : // 外部接口调用Request()
240 0 : HcclResult Request() {
241 0 : CHK_SMART_PTR_NULL(retryBase_);
242 0 : return retryBase_->Handle(this);
243 : }
244 :
245 0 : u32 GetRankId() {
246 0 : return rankId_;
247 : }
248 :
249 18 : const char *GetOpRetryMachineType() const {
250 18 : std::string ctxType = isRootRetryCtx_ ? "Server" : "Agent";
251 36 : return ctxType.c_str();
252 18 : }
253 :
254 0 : bool IsRootRetryCtx() {
255 0 : return isRootRetryCtx_;
256 : }
257 :
258 0 : const char *GetDfxIpInfo() const {
259 0 : return localRetryInfo_.dfxIpInfo;
260 : }
261 :
262 0 : void ResetAgentState () {
263 0 : localRetryInfo_.opInfo.execStatus.kfcError = KfcError::kNone;
264 0 : localRetryInfo_.isNeedReportOpRetryErr = false;
265 0 : isBSRRdmaRecvError_ = false;
266 0 : isBSRRdmaSendError_ = false;
267 0 : }
268 :
269 0 : void ResetServerState () {
270 0 : errorRankList_.clear();
271 0 : needRetryServerRanks_.clear();
272 0 : isNeedReportOpRetryErr = false;
273 0 : }
274 :
275 3 : std::shared_ptr<HDCommunicate> GetH2dPtr() {
276 3 : return h2dPtr_;
277 : }
278 :
279 2551563 : std::shared_ptr<HDCommunicate> GetD2hPtr() {
280 2551563 : return d2hPtr_;
281 : }
282 :
283 0 : bool IsPaused() const {
284 0 : return isPaused_;
285 : }
286 :
287 : std::string group_ = "";
288 : s32 deviceLogicId_ = INVALID_INT;
289 : u32 rankId_ = INVALID_UINT;
290 : bool haveCommEnableBackupLink_ = false;
291 :
292 : // agent状态机储存信息
293 : std::shared_ptr<HcclSocket> agentSocket_ = nullptr;
294 : std::shared_ptr<HcclOpStreamRes> opStreamPtr_ = nullptr;
295 : OpRetryResetNotifyCallback notifyResetCallback_ = nullptr;
296 : OpRetrySetTransportStatusCallback setTransportStatusCallback_ = nullptr;
297 : OpRetryGetSwitchRanksCallback getSwitchRanksCallback_ = nullptr;
298 : OpRetrySetTransportResumeStatusCallBack setTransportReseumeStatusCallback_ = nullptr;
299 : bool isEnableBackupLink_ = false;
300 : bool isEnableSdmaRetry_ = false;
301 : RetryInfo localRetryInfo_;
302 : ChangeLinkInfo localChangeLinkInfo_;
303 : LinkPortStatus linkPortStatus_;
304 : bool isChangeLinkInfoInit_ = false;
305 : std::map<u32, bool> lastLinkPortStatus_;
306 : bool isUseDefaultPort_ = true;
307 : HcclNetDevCtx netDevCtx_ = nullptr;
308 : HcclNetDevCtx backUpNetDevCtx_ = nullptr;
309 : bool isBSRRdmaRecvError_ = false;
310 : bool isBSRRdmaSendError_ = false;
311 : HcclOpIdentifier RemainSendOpId_;
312 : HcclOpIdentifier RemainRecvOpId_;
313 : ActiveSwitchInfo switchInfo_;
314 : bool isAgentStateWaitResume_ = false;
315 :
316 : bool isRecivedCmdToRunning = false;
317 : bool isRecivedCmdToCheckLink = false;
318 : // server状态机储存信息
319 : std::map<u32, HcclAgentRetryInfo> serverSockets_;
320 : std::vector<u32> needRetryServerRanks_;
321 : HcclOpIdentifier curFaultOpId;
322 : std::map<u32, HcclOpIdentifier> errorRankList_;
323 : bool isRdmaError = false;
324 : bool isAlreadyChangeLink = false;
325 : std::map<u32, ActiveSwitchInfo> switchInfoMap_;
326 : bool isServerStateWaitResume_ = false;
327 : bool isNeedReportOpRetryErr = false; // 针对重执行算子不一致和inplace场景，上报故障
328 :
329 : bool isOpRetryQuit = false;
330 : bool isPaused_ = false;
331 : private:
332 : std::shared_ptr<OpRetryBase> retryBase_ = nullptr;
333 : RetryState state_ = RETRY_STATE_RESERVED;
334 : bool isRootRetryCtx_ = false;
335 :
336 : std::shared_ptr<HDCommunicate> h2dPtr_ = nullptr;
337 : std::shared_ptr<HDCommunicate> d2hPtr_ = nullptr;
338 : };
339 : }
340 : #endif
|
