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