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 __AICPU_COMMUNICATOR_H__
12 : #define __AICPU_COMMUNICATOR_H__
13 :
14 : #include <memory>
15 : #include <vector>
16 : #include <iterator>
17 : #include <array>
18 : #include <hccl/hccl_types.h>
19 : #include "common/aicpu_hccl_def.h"
20 : #include "log.h"
21 : #include "mem_device_pub.h"
22 : #include "stream_pub.h"
23 : #include "local_notify.h"
24 : #include "comm_factory_pub.h"
25 : #include "coll_executor_base.h"
26 : #include "dispatcher.h"
27 : #include "coll_alg_param.h"
28 : #include "transport_pub.h"
29 : #include "hccl_common.h"
30 : #include "aicpu_operator_pub.h"
31 : #include "peterson_lock.h"
32 : #include "aicpu_hdc.h"
33 : #include "aicpu_zero_copy_exchanger.h"
34 : #include "cann_error_reporter.h"
35 : #include "hccl_trace_info.h"
36 : #include "aicpu_share_data_manager.h"
37 : #include <shared_mutex>
38 : #include "hccl/hccl_res.h"
39 : #include "channel_param.h"
40 : #include "aicpu_launch_manager.h"
41 : #include "aicpu_ts_thread.h"
42 : #include "new/hccl_dispatcher_ctx.h"
43 : #include "aicpu_init_param.h"
44 : #include "task_exception.h"
45 : #include "ub_transport_lite_impl.h"
46 : #include "aicpu_cache_manager.h"
47 :
48 : namespace hccl {
49 :
50 : enum class CommTransportsType {
51 : INVALID,
52 : GENERAL, // 通信域通用，input和output的中转内存全部使用cllbuffer
53 : SPECIAL, // 通信域专用，input和output的中转内至少其中一个不是cllbuffer
54 : };
55 :
56 : enum class CommResourceCtrlType {
57 : INVALID,
58 : INIT, // 资源初始化
59 : REFRESH, // 资源刷新
60 : };
61 :
62 : struct TagAddress {
63 : u64 addr0;
64 : u32 key0 = 0;
65 : u64 addr1;
66 : u32 key1 = 0;
67 : // 重载运算符==
68 : bool operator==(const TagAddress &other) const
69 : {
70 : return addr0 == other.addr0 && addr1 == other.addr1 && key0 == other.key0 && key1 == other.key1;
71 : }
72 : };
73 :
74 : struct RankData {
75 : u32 remoteWorldRank{INVALID_VALUE_RANKID};
76 : u32 remoteUsrRankId{INVALID_VALUE_RANKID};
77 : };
78 :
79 : enum class CqeExceptionStatus : uint32_t {
80 : kNone = 0,
81 : kSdmaErr, // 可重执行的ErrCqe
82 : kOther // 其他ErrCqe
83 : };
84 : enum class AicpuKfcHandlerType: u32 {
85 : kSetStepSize,
86 : kNotifyRecord,
87 : kNotifyWait,
88 : kClearMsgArea,
89 : kClearCommitTurn,
90 : kSetProfTimeStart,
91 : kSetProfTimeOrch,
92 : kSetProfTimeEnd,
93 : kMax
94 : };
95 :
96 : struct AicpuStreamMontior {
97 : HcclUs historyTime;
98 : u32 historyHead;
99 : u32 historyTaskId;
100 : u32 historyType;
101 : };
102 : using AicpuKfcHandler = std::function<HcclResult(const std::vector<u64> &)>;
103 : class HcclCommAicpu {
104 : public:
105 : explicit HcclCommAicpu();
106 : ~HcclCommAicpu();
107 : HcclResult Init(const HcclOpResParam *commParam, bool isCustom);
108 34 : Stream &GetMainStream() { return mainStream_; }
109 4 : std::vector<Stream> &GetSlaveStream() { return slaveStreams_; }
110 17 : HcclDispatcher GetDispatcher() const { return dispatcher_; }
111 30 : const std::string &GetGroupName() const { return identifier_; }
112 : void SetAlgType(u64 algType);
113 : void SetDebugMode(u8 debugMode);
114 : void SetSendRecvInfoPtr(void* sendRecvInfoPtr);
115 : void SetDumpDebug(bool dumpDebug);
116 15 : void SetIsDeviceMode(bool isDeviceMode) { isDeviceMode_ = isDeviceMode; }
117 0 : void SetUserStreamId(s32 userStreamId) { userStreamId_ = userStreamId; }
118 : HcclResult StreamTaskMonitor(void);
119 : HcclResult UpdateNotifyWaitTimeOut(SyncMode syncMode, u64 notifyWaitTime);
120 : HcclResult GetStreamAll(std::vector<Stream> &streams);
121 59 : u32 GetDevId() const { return devId_; }
122 57 : DfxExtendInfo *GetDfxExtendInfo() { return &dfxExtendInfo_; }
123 25 : DevType GetDevType(void) const { return topoInfo_.deviceType; }
124 19 : uint32_t GetRankSize(void) const { return topoInfo_.userRankSize; }
125 : HcclResult ExecOp(const std::string &newTag, const std::string &algName, OpParam &opParam,
126 : const HcclOpResParam *commParam);
127 : HcclResult GetAlgResponseRes(const std::string &newTag, const std::string &algName, const OpParam &opParam,
128 : const HcclOpResParam *commParam, std::unique_ptr<CollExecutorBase> &executor,
129 : AlgResourceResponse *&algResResponse);
130 : void PrepareOpRetryHandler(u8 inplaceSupportRetry, u8 retryEnable, u8 inPlaceSupportRetryStatus,
131 : u8 isInplacePreSync, u8 isPostSync);
132 : void NsCommStop();
133 : void NsCommClean();
134 15 : void SetAicpuRpcServer(u64 rpc) { rpc_ = rpc; }
135 : HcclResult GetSuspendingFlag(HcclComSuspendingFlag &flag);
136 : HcclResult BackGroundGetCmd(KfcCommand &cmd);
137 : HcclResult BackGroundSetStatus(KfcStatus status);
138 25 : void SetNsOpStatus(bool status) { isOpLaunch = status; }
139 0 : bool BackGroundGetOpStatus() { return isOpLaunch; }
140 3 : void SetNsStopLaunchStatus(bool status) { endStopLaunch = status; }
141 32 : bool GetNsStopLaunchStatus() { return endStopLaunch; }
142 11 : void SetCommInfoStreamStatus(bool status) { groupNsCommStatus_ = status; }
143 0 : bool GetCommInfoStreamStatus() const { return groupNsCommStatus_; }
144 0 : bool GetCommInfoStatus() const { return commOpenStatus; }
145 : HcclResult GetBackGroundCommand(BackgroundCommand &bgCmd);
146 : HcclResult ResponseBackGroundStatus(KfcExecStatus &status);
147 : HcclResult GetKfcCommand(KfcCommand &cmd);
148 2 : void SetCommRecoveryFlag(bool status) { commNeedsRecovery = status; }
149 0 : bool GetCommRecoveryFlag() { return commNeedsRecovery; }
150 : void RecordReportStatus(dfx::ReportStatus status);
151 : void GetReportStatusQueue(std::queue<dfx::ReportStatus> &reportStatusQue);
152 : HcclResult Orchestrate(const std::string &newTag, const std::string &algName, OpParam ¶m, std::unique_ptr<CollExecutorBase> &executor,
153 : AlgResourceResponse &algResource, const HcclOpResParam *commParam);
154 : HcclResult SaveTraceInfo(std::string &logInfo);
155 : HcclResult FlushUtraceInfo();
156 : std::string GetExcuteOp();
157 : void HandleCqeException(hccl::Stream &stream, bool isReadClear);
158 : void HandleIndOpCqe();
159 : void ReportIndOpCqe(hccl::Stream &stream, const rtLogicCqReport_t &cqeException, CqeStatus cqeStatus);
160 140 : static void ResetErrMsgReport() { errMessageReport_ = true; };
161 : void PrintTaskExceptionAllComm();
162 : HcclResult PrintTaskExceptionAllThreads();
163 : bool GetOpRetryEnable();
164 : void SetZeroCopyEnable(bool enable);
165 : void SetSymmetricMemoryEnable(bool enable);
166 : bool IsTaskExceptionForHccs();
167 : u32 HcclGetWaitStopExecCmdTimeout();
168 : u32 HcclGetWaitRetryCmdTimeout(uint32_t retryCnt);
169 : HcclResult UpdateOpExecStatus(HcclOpExecFSM &fsmState, KfcStatus state, KfcError &errorCode, uint32_t retryCnt);
170 : HcclResult ResetOpRetryException(HcclCMDType opType);
171 : HcclResult CleanAllRoceResource();
172 : // aclgraph 销毁时通过 KFC 投递清理 aicpu 端单一 tag 的全部关联资源，配合 host 端 AclgraphCallback 使用
173 : HcclResult ClearOpResource(const std::string &tag);
174 : HcclResult SwitchNic();
175 : HcclResult ResumeChangeLink();
176 : HcclResult ParseHierarchicalAlgOption(u32 *ahcConfInfo);
177 120 : void RegisterKfcHandler(AicpuKfcHandlerType type, AicpuKfcHandler cb) { kfcHandlers_[static_cast<size_t>(type)] = cb; }
178 : HcclResult RecordHostOrder(const HcclOpResParam *commParam, const std::string& tag, u8 orderLaunchMode); // kernel占到核后，通知host侧
179 : std::string GetTaskExceptionTaskInfo(u32 sqHead, SqeRingBuffer *sqeContextBuffer, uint8_t &type, uint16_t &taskId, uint32_t &remoteRank);
180 : HcclResult RegisterProfCallBack();
181 : // 独立算子专用
182 : HcclResult SetChannelP2pNotify(TransportDeviceP2pData &transDevP2pData, u64 &p2pNotifyNum,
183 : HcclChannelP2p &channelP2p);
184 : HcclResult SetChannelRoceNotify(TransportDeviceIbverbsData &transDevIbverbsData, u64 &roceNotifyNum,
185 : HcclChannelRoce &channelRoce);
186 : HcclResult InitP2pChannel(HcclIndOpChannelRemoteResV3 *commParam, uint32_t channelIndex);
187 : HcclResult InitRoceChannel(HcclIndOpChannelRemoteResV3 *commParam, uint32_t channelIndex);
188 : HcclResult AllocChannelResource(HcclIndOpChannelRemoteResV3 *commParam);
189 :
190 : HcclResult InitAicpuIndOp(CommAicpuParam *commAicpuParam);
191 0 : bool GetIsInitIndOp() { return indOpCommInitialized_; };
192 : HcclResult InitThreads(ThreadMgrAicpuParam *param);
193 : HcclResult NotifyFree(NotifyMgrAicpuParam *param);
194 : HcclResult NotifyAlloc(NotifyMgrAicpuParam *param);
195 :
196 : HcclResult RegisterOpInfo(void* opInfo, u32 size);
197 : HcclResult RegOpTaskException(HcommGetOpInfoCallback callback);
198 : HcclResult InitProfthreadResource(u32 threadNum);
199 :
200 : HcclResult SetDispatcherCtxOnThread();
201 : private:
202 : HcclResult SetHrtWorkMode(const HcclOpResParam *commParam);
203 : HcclResult SetHrtDeviceSatMode(const HcclOpResParam *commParam);
204 : HcclResult InitSlaveStreamObjs(const HcclOpResParam *commParam);
205 : HcclResult InitLocalNotifyObj(const HcclOpResParam *commParam);
206 : HcclResult InitOpNotifyObj(const HcclOpResParam *commParam);
207 : HcclResult StreamRestore(u32 streamId); // 将流资源映射到custom进程
208 : HcclResult ParseTlvToVector(u64 srcTlv, u64 srcTlvTotalLength,
209 : std::vector<std::vector<std::vector<u32>>> &vectorInfo);
210 : HcclResult ParseTlvToSubGroupVector(u64 srcTlv, u64 srcTlvTotalLength,
211 : std::vector<std::vector<std::vector<std::vector<u32>>>> &vectorInfo);
212 : HcclResult InitLocalTagRes(const ListCommon &head, bool reAllocFlag = false);
213 : HcclResult InitRemoteTagRes(u32 &rankId, const ListCommon &head, const std::string &newTag, u32 notifyNum,
214 : TransportLinkType linkType = TransportLinkType::RDMA);
215 : template <typename T>
216 : HcclResult InitAndVerifySignal(const HcclSignalInfo &signalInfo, std::vector<std::shared_ptr<T>> ¬ifyVec);
217 : HcclResult InitTopoMatcher();
218 : HcclResult InitTopoInfo(const HcclOpResParam *commParam);
219 : HcclResult InitCclbuffer(const HcclOpResParam *commParam);
220 : HcclResult InitConfigInfo(const HcclOpResParam *commParam);
221 : HcclResult InitMainStreamObj(const HcclOpResParam *commParam);
222 : HcclResult InitOrderStreamObj(const HcclOpResParam *commParam);
223 : HcclResult InitStreamObj(const HcclStreamParam& streamParam, Stream& stream);
224 : HcclResult InitTimeOutConfig(const HcclOpResParam *commParam);
225 : HcclResult InitHostDeviceLock(const HcclOpResParam *commParam);
226 : HcclResult InitOpRetry(const HcclOpResParam *commParam);
227 : HcclResult InitZeroCopyExchanger(const HcclOpResParam *commParam);
228 : HcclResult PrepareZeroCopyExchanger(const std::string &newTag, OpParam &opParam,
229 : AlgResourceResponse *algResResponse);
230 : HcclResult RegisterDispatcherCallback();
231 : HcclResult RegisterProfilingCallback();
232 : HcclResult InitUtraceInfo(const HcclOpResParam *commParam);
233 : void InitSendRecvOpId(const OpParam ¶m, HcclOpIdentifier &opId);
234 : HcclResult GetStreamData(
235 : const HcclStreamInfo &streamInfo, HcclComStreamInfo &comStreamInfo, u32 &sqHead, u32 &sqTail);
236 : HcclResult RefreshTransportsResForRank(const HcclOpResParam *commParam, u32 rankId,
237 : const std::string &newTag, u32 notifyNum, TransportLinkType linkType = TransportLinkType::RDMA);
238 : HcclResult GetRdmaLinksByRankAndTag(const HcclOpResParam *commParam, CommTransportsType type, u32 rankId,
239 : const std::string &newTag, LINK &link, bool isBackup, u32 notifyNum, bool isSecond);
240 : HcclResult GetSdmaLinksByRankAndTag(const HcclOpResParam *commParam, CommTransportsType type, u32 rankId,
241 : const std::string &newTag, LINK &link, bool isBackup, u32 notifyNum,
242 : TransportLinkType linkType = TransportLinkType::RESERVED);
243 : HcclResult AllocTransportResource(const std::string &newTag, const OpParam &opParam,
244 : const HcclOpResParam *commParam, AlgResourceRequest &resRequest, AlgResourceResponse &algResResponse);
245 : HcclResult IncreAllocTransportResource(const std::string &newTag, const OpParam &opParam,
246 : const HcclOpResParam *commParam, AlgResourceRequest &resRequest, AlgResourceResponse &algResResponse);
247 : HcclResult CreateLink(const std::string &newTag, TransportRequest& transportRequest,
248 : const HcclOpResParam *commParam, LINK& link, u32 notifyNum, bool isBackup, bool isSecond = false);
249 : HcclResult AllocLocalNotifysResource(const std::string &newTag, const HcclOpResParam *commParam,
250 : const u32 notifyNum, std::vector<std::shared_ptr<LocalNotify>> ¬ifiesMain,
251 : std::vector<std::shared_ptr<LocalNotify>> ¬ifiesAux);
252 : HcclResult AllocStreamsResource(
253 : const std::string &newTag, const HcclOpResParam *commParam, const u32 streamNum, std::vector<Stream> &streams);
254 : HcclResult AllocScratchMemResource(const std::string &newTag, const HcclOpResParam *commParam,
255 : const u64 &scratchMemSize, DeviceMem &scratchMem, bool reAllocFlag = false);
256 : HcclResult AllocAlgResource(const std::string &newTag, const OpParam &opParam, const HcclOpResParam *commParam,
257 : AlgResourceRequest &resRequest, AlgResourceResponse &algResResponse);
258 : HcclResult CalcResRequest(const std::string &algName, const OpParam ¶m,
259 : std::unique_ptr<CollExecutorBase> &executor, AlgResourceRequest &resourceRequest);
260 : HcclResult WaitFinishWhileLoop(Stream &mainStream, std::vector<Stream> &subStreams, std::string &tag,
261 : const uint32_t &beginSqePos, OpParam ¶m);
262 : HcclResult CheckOpExecStatusCallback();
263 : HcclResult CheckOpExecStatus();
264 : HcclResult UpdateSuspendStatus(const OpParam ¶m, HcclOpExecFSM &fsmState, KfcError &errorCode, uint32_t retryCnt);
265 : HcclResult CheckTaskTimeout(const Stream &mainStream, const uint64_t startUsec);
266 :
267 : HcclResult HcclOpExecFsmInitProcess(const std::string &newTag, OpParam ¶m, AlgResourceResponse &algResource,
268 : HcclOpExecFSM &fsmState, KfcError &errorCode);
269 : bool HcclOpCheckSupportRetry(HcclCMDType opType);
270 : HcclResult HcclOpExecChangeLinkProcess(const std::string &newTag, HcclOpExecFSM &state, KfcError &errorCode,
271 : uint32_t &retryCnt, AlgResourceResponse &algResource, const HcclOpResParam *commParam, const OpParam ¶m);
272 : HcclResult HcclOpExecFsmLaunchProcess(const std::string &algName, OpParam ¶m,
273 : std::unique_ptr<CollExecutorBase> &executor, AlgResourceResponse &algResource, HcclOpExecFSM &fsmState,
274 : KfcError &errorCode, uint32_t &beginSqePos, uint32_t &endSqePos, uint32_t retryCnt);
275 : HcclResult HcclOpExecFsmWaitEndProcess(OpParam ¶m, AlgResourceResponse &algResource, HcclOpExecFSM &fsmState,
276 : KfcError &errorCode, uint32_t retryCnt, std::string &tag, const uint32_t &beginSqePos);
277 : HcclResult HcclOpExecFsmStoppingProcess(const OpParam ¶m, HcclOpExecFSM &fsmState, KfcError &errorCode, uint32_t retryCnt);
278 : HcclResult HcclOpExecFsmStoppedProcess(HcclOpExecFSM &fsmState, KfcError &errorCode, uint32_t retryCnt,
279 : const std::string &algName, OpParam ¶m, uint32_t beginSqePos, uint32_t endSqePos);
280 : HcclResult HcclOpExecFsmWaitRetryProcess(const OpParam ¶m, HcclOpExecFSM &fsmState, KfcError &errorCode, KfcCommand &lastCmd);
281 : HcclResult ResetSqBuff();
282 : HcclResult CleanStreamFunc();
283 : HcclResult UpdateSqStatus(Stream &stream);
284 : HcclResult HcclOpExecFsmRetryProcess(const std::string &algName, OpParam ¶m,
285 : std::unique_ptr<CollExecutorBase> &executor, AlgResourceResponse &algResource, HcclOpExecFSM &fsmState,
286 : KfcError &errorCode, uint32_t &retryCnt, uint32_t &beginSqePos, uint32_t &endSqePos);
287 : HcclResult RetryOrchestrateHcclOp(const std::string &algName, OpParam ¶m,
288 : std::unique_ptr<CollExecutorBase> &executor, AlgResourceResponse &algResource, uint32_t &beginSqePos,
289 : uint32_t &endSqePos);
290 : HcclResult HcclOpExecFsmEndProcess(uint32_t retryCnt);
291 : std::string PrintInplaceSupportRetryStatus(InplaceSupportRetryStatus inPlaceSupportRetryStatus);
292 : bool HcclOpSupportRetry(const std::string &algName, bool retryEnable, OpParam ¶m);
293 : std::string PrintInplaceStatus(u8 isInplaceStatus);
294 : HcclResult SupportRetryWithInplaceCheck(const std::string &algName, OpParam ¶m);
295 : bool isPollutedZeroCopyOp(OpParam ¶m);
296 : bool HcclOpCheckNsRecovery();
297 : HcclResult OrchestrateHcclOp(const std::string &algName, OpParam ¶m,
298 : std::unique_ptr<CollExecutorBase> &executor, AlgResourceResponse &algResource, uint32_t &beginSqePos,
299 : uint32_t &endSqePos);
300 : HcclResult LaunchSlaveStreamTask(AlgResourceResponse &algResource);
301 : HcclResult GetAlltoAllvSendRecvInfo(const void* sendRecvInfoPtr, HcclDataType sendType, HcclDataType recvType);
302 : HcclResult GetAlltoAllvcSendRecvInfo(const void *sendCountMatrix, HcclDataType sendType, HcclDataType recvType);
303 : HcclResult CheckSendRecvParams(const std::vector<SendRecvInfo> &allMeshAggregationSendRecvInfo);
304 : HcclResult SetAlltoAllInputAndOutPutMem(OpParam ¶m, AlgResourceResponse &algResource);
305 : HcclResult NotifyPost(void);
306 : HcclResult NotifyWait(void);
307 : HcclResult GetAlltoAllTotalCount(OpParam ¶m, u64 &sendCount, u64 &recvCount);
308 : HcclResult GetAlltoAllVTotalCount(OpParam ¶m, u64 &sendCount, u64 &recvCount);
309 : HcclResult GetAlltoAllVCTotalCount(OpParam ¶m, u64 &sendCount, u64 &recvCount);
310 :
311 : // taskException
312 : void PollCqeException(hccl::Stream &stream, bool isReadClear, rtLogicCqReport_t &cqeException, CqeStatus &cqeStatus);
313 : void ExchangeCqeContext(hccl::Stream &stream, rtLogicCqReport_t &cqeException, CqeStatus &cqeStatus,
314 : ErrCqeContext &cqeCtx);
315 : void ReportErrCqe(hccl::Stream &stream, ErrCqeContext &cqeCtx);
316 : HcclResult PrintTaskExceptionAllStreams();
317 : bool IsRepeatedOpTaskException(u32 idx, SqeRingBuffer *sqeContextBuffer); // 避免同一个算子重复打印taskException
318 : std::string GetTaskExceptionOpInfo(u32 idx, SqeRingBuffer *sqeContextBuffer); // 打印算子参数信息
319 : void PrintTaskExceptionTaskQue(u32 sqIdx, SqeRingBuffer *sqeContextBuffer, bool isMonitor = false); // 打印当前位置的前序task
320 : std::string GetTaskBriefsInfo(u32 idx, SqeRingBuffer *sqeContextBuffer); // 打印task简写
321 : void PrintAicpuCommExecStatus();
322 :
323 : HcclResult UpdateOpRingBufferIdx();
324 : HcclResult CombineReportOpInfo(OpParam ¶m, bool isRetry, bool isRelay);
325 : void UpdateBSRRetryCnt();
326 : void ResetBSRRetryCnt();
327 : HcclResult CommitBSRStoredException(HcclOpExecFSM &fsmState, KfcError &errorCode);
328 : HcclResult QueryBatchSendRecvPairBeginPos();
329 : HcclResult QueryBatchSendRecvPairEndPos();
330 : HcclResult UpdateOpExecStatus(HcclOpExecFSM &fsmState, HcclOpIdentifier &opId, KfcStatus state,
331 : KfcError &errorCode, uint32_t retryCnt);
332 : u32 HcclUpdateBatchSendRecvOpIndex(std::map<u32, u32> &bsrIndexMap, u32 peerRank);
333 : u32 HcclUpdateBatchSendRecvOpIndex(HcclSendRecvType opType, u32 srcRank, u32 dstRank);
334 : HcclResult InitBatchSendRecvOpId(const OpParam ¶m, const HcclSendRecvItem* sendrecvPair,
335 : HcclOpIdentifier &opId, u32 streamId, AlgResourceResponse &algResource);
336 : HcclResult InitBatchSendRecvOpId(const OpParam ¶m, AlgResourceResponse &algResource);
337 : HcclResult InitBsrSendRecvOpIdAndExcuteOpId(OpParam ¶m, AlgResourceResponse &algResource,
338 : HcclOpExecFSM &fsmState, KfcError &errorCode);
339 : void SetBSRSendOpExecException();
340 : void SetBSRRecvOpExecException();
341 : bool GetBSRSendOpExecException();
342 : bool GetBSRRecvOpExecException();
343 :
344 : HcclResult CleanStream(Stream &stream);
345 : HcclResult ClearStreamCqeException(Stream &stream);
346 : HcclResult ResetBSRSendOpExecException();
347 : HcclResult ResetBSRRecvOpExecException();
348 : HcclResult ResetBSRException();
349 : HcclResult BSRStopedProcess(HcclOpExecFSM &fsmState, KfcError &errorCode);
350 : HcclResult GetBSRRetryOpId(const OpParam ¶m, HcclOpIdentifier &targetOpId);
351 : HcclResult InitExecLoop(OpParam ¶m, std::unique_ptr<CollExecutorBase> &executor, u32 &loopNum);
352 : template <typename T>
353 : HcclResult InitAndVerifySingleSignal(const HcclSignalInfo &signalInfo, std::shared_ptr<T> ¬ify);
354 : HcclResult SetTransportMachinePara(MachinePara &machinePara, u32 &rankId, const std::string &newTag,
355 : TransportLinkType linkType = TransportLinkType::RESERVED);
356 : HcclResult CheckNotifyOrQPMaxNum(u64 &existNum, const u64 &MaxNum, const bool &isNotifyRes);
357 : HcclResult SetTagRemoteRes(u32 &rankId, const std::string &tag, HccltagRemoteResV2 *tagRes);
358 : HcclResult SetTransportPtpNotify(TransportDeviceP2pData &transDevP2pData,
359 : u64 &p2pNotifyNum, HcclLinkP2pV2 &linkP2p, u32 notifyNum);
360 : HcclResult SetTransportRoceQP(TransportDeviceIbverbsData &transDevIbverbsData,
361 : u64 &roceQpNum, HcclLinkRoceV2 *linkRoce);
362 : HcclResult SetTransportRoceNotify(TransportDeviceIbverbsData &transDevIbverbsData,
363 : u64 &roceNotifyNum, HcclLinkRoceV2 *linkRoce, u32 notifyNum);
364 : HcclResult InitLinkP2p(HccltagRemoteResV2 *tagRes, u32 &rankId, const std::string &newTag, u32 notifyNum,
365 : TransportLinkType linkType = TransportLinkType::RESERVED);
366 : HcclResult InitLinkRoce(HccltagRemoteResV2 *tagRes, u32 &rankId, const std::string &newTag, u32 notifyNum,
367 : const bool isBackup = false);
368 : HcclResult InitLinkRoce(HccltagRemoteResV2 *tagRes, HcclLinkRoceV2 *linkRoce, u32 &rankId,
369 : const std::string &newTag, u32 notifyNum, const bool isBackup = false, const bool isSecond = false);
370 : HcclResult GetBsrTransportQpn( const HcclSendRecvItem *sendrecvPair, AlgResourceResponse &algResource,
371 : u32 &qpn);
372 : HcclResult ReAllocTransportResource(const std::string &newTag, AlgResourceResponse &algResResponse,
373 : std::map<u32, bool> &remoteRankPortMap, const HcclOpResParam *commParam, const OpParam ¶m);
374 : HcclResult CleanRoceResource(const std::string &newTag, AlgResourceResponse &algResResponse,
375 : const std::map<u32, bool> &remoteRankPortMap, const OpParam ¶m);
376 : HcclResult LoadChangeLinkInfo(ChangeLinkInfo &changeLinkInfo);
377 :
378 : HcclResult AddRetryExecFlipTask(AlgResourceResponse &algResource);
379 : HcclResult ReportHcclTaskInfo(Stream &mainStream, std::vector<Stream> &subStreams);
380 : HcclResult ClearLocalBuff(Stream &mainStream, std::vector<Stream> &subStreams);
381 : HcclResult UpdateProfReportStartSqeIdx();
382 : HcclResult TasktypeTransferD2H(const uint8_t sqeType, TaskType &taskType);
383 : void PrepareMc2Handler();
384 : HcclResult InitOpCounter(const OpCounterInfo &opCounterInfo);
385 :
386 : // rts调用接口,通过mailbox上报给tsfw
387 : HcclResult SendTaskExceptionByMBox(const uint16_t &rsErrorCode);
388 :
389 : HcclResult RefreshLinkForSwitchNic(const std::string &newTag, const TransportRequest &transportRequest,
390 : const std::map<u32, bool> &remoteRankPortMap, bool isSecondBuild, LINK &switchLink);
391 : HcclResult ReAllocTransportForSwitchNic(const std::string &newTag, AlgResourceResponse &algResResponse,
392 : std::map<u32, bool> &remoteRankPortMap);
393 : HcclResult RefreshRoceTransportsForSwitchNic(std::unordered_map<std::string, OpCommTransport> &reservedLinks);
394 : HcclResult RevertTransportsForSwitchNic(std::unordered_map<std::string, OpCommTransport> &reservedLinks);
395 : HcclResult SwitchNicWaitHandleCommand(std::unordered_map<std::string, OpCommTransport> &reservedLinks);
396 : HcclResult SwitchNicWaitResult(std::unordered_map<std::string, OpCommTransport> &reservedLinks);
397 : u32 CalculateOpExecIndex(const OpParam &opParam, u32 userRank); // 每次展开时计算
398 :
399 : HcclResult InitProfResource();
400 : void InitCommInfoStatus(bool commInfo);
401 : HcclResult InitTinyMem(const HcclOpResParam *commParam);
402 : HcclResult SetStreamEnable(Stream &stream);
403 : HcclResult RefreshAlgResponseTransportRes(const std::string &newTag, AlgResourceResponse& algResResponse,
404 : std::map<u32, bool> &remoteRankPortMap, bool isChangeLinkFlag,
405 : const HcclOpResParam *commParam, const OpParam ¶m);
406 : HcclResult RefreshCommResponseTransportRes(std::map<u32, bool> &remoteRankPortMap);
407 : HcclResult PrintTaskExceptionByTaskId(u8 sqeType, u16 taskId, hccl::Stream &stream, u32 tail);
408 : bool IsNoNeedWait(void);
409 : void SetStreamCqeExceptionStatus(const Stream &stream, CqeExceptionStatus cqeStatus);
410 : void ResetStreamCqeExceptionStatus(const Stream &stream);
411 : CqeExceptionStatus GetStreamCqeExceptionStatus(const Stream &stream);
412 : HcclResult GenTaskExceptionInfo(u8 sqeType, hccl::Stream &stream, u32 head);
413 : HcclResult InvokeKfcHandler(AicpuKfcHandlerType type, const std::vector<u64> args);
414 :
415 : bool IsNoNeedMonitor(void);
416 : void InsertMonitorData(Stream &stream, HcclUs &curTime, u32 sqHead, uint16_t taskId, uint8_t type);
417 : bool IsNeedRefreshMonitorData(AicpuStreamMontior &streamMontior, HcclUs &curTime, uint32_t remoteRank,
418 : uint16_t taskId, u32 sqHead, u32 sqTail, uint8_t type);
419 : //对称内存
420 : HcclResult PrepareSymmetricMemory(const OpParam ¶m, OpCommTransport &opTransportResponse);
421 : HcclResult PrepareSymmetricMemRanges(const AlgResourceResponse &algResource, uint64_t inputSize, uint64_t outputSize,
422 : std::vector<OpUnfoldMemRange>& userInputMemRanges, std::vector<OpUnfoldMemRange>& userOutputMemRanges);
423 :
424 : HcclResult CalSendRecvInfoForAlltoall(const OpParam ¶m);
425 : HcclResult CalSendRecvInfoFor910B(const std::string &algName, const OpParam ¶m,
426 : std::unique_ptr<CollExecutorBase> &executor);
427 : void HandleExistTagReAlloc(HccltagLocalResV2* tagRes, const std::string& tag, bool reAllocFlag,
428 : ListCommon*& curList, bool& needSkip);
429 :
430 : // 获取指定索引的算子信息
431 : const AicpuOpInfo* GetOpInfoFromSqIdx(u32 sqIdx, SqeRingBuffer *sqeContextBuffer);
432 :
433 : // 打印算子数据信息
434 : void PrintOpDataInfo(u32 sqIdx, SqeRingBuffer *sqeContextBuffer, bool isMonitor);
435 :
436 : // 打印task序列行
437 : void PrintTaskLine(bool isMonitor, u32 lineNum, u32 totalPrinted, const std::string& taskLine) const;
438 :
439 : // 更新算子上下文
440 : void UpdateOpContext(u32& opIndex, std::string& opTag, u32& lineCount, std::vector<std::string>& currentOpTasks, u32 newOpIndex, const std::string& newOpTag) const;
441 :
442 : // 准备下一行数据
443 : void PrepareNextLine(u32 opIndex, u32& lineCount, std::vector<std::string>& currentOpTasks) const;
444 :
445 : // 拼接task列表为字符串
446 : std::string ConcatTaskLine(const std::vector<std::string>& tasks) const;
447 :
448 : // 打印剩余未满行的tasks
449 : void PrintRemainingTasks(bool isMonitor, u32 lineCount, u32 printedCount, const std::vector<std::string>& currentOpTasks) const;
450 :
451 : std::unordered_map<s32, u32> opExecIndexMap_;
452 :
453 : // 管理aicpu和custom进程共享的数据
454 : AicpuShareDataManager aicpuShareData_;
455 : bool isCustom_ = false;
456 :
457 : // local资源
458 : std::vector<Stream> slaveStreams_;
459 : Stream mainStream_;
460 : Stream orderStream_;
461 : std::unordered_set<u32> streamToObj_; // 从context的资源构造为Stream对象去重
462 : s32 userStreamId_; // 用户传入的stream的id
463 :
464 : std::vector<std::shared_ptr<LocalNotify>> localNotifies_; // 主从流之间同步的notify
465 : std::vector<std::shared_ptr<LocalNotify>> opNotifies_; // host与device间同步的notify
466 : std::vector<std::shared_ptr<LocalNotify>> orderNotifies_{AICPU_ORDER_NOTIFY_MAX_NUM, nullptr}; // 按序下发的notify
467 : std::unordered_set<u32> notifysToObj_; // 从context的资源构造为LocalNotify对象去重
468 :
469 : std::unordered_map<std::string, std::shared_ptr<DeviceMem>> tagScratchMem_; // 本地scratchmem
470 : std::unordered_map<std::string, std::unordered_set<u64>>
471 : localTagResToObj_; // 从context的localtag资源构造为对象去重
472 :
473 : // 跨卡资源
474 : uint32_t notifySize_;
475 : const HcclOpResParam *commParam_ = nullptr;
476 :
477 : // 通信域内的link
478 : std::unordered_map<u32, std::unordered_map<std::string, std::shared_ptr<Transport>>>
479 : linkRes_; // 通信域内的SDMA hccs链路，包括通用和专用
480 : // 通信域内的SDMA sio链路
481 : std::unordered_map<u32, std::unordered_map<std::string, std::shared_ptr<Transport>>> linkResSio_;
482 : std::unordered_map<u32, std::unordered_map<std::string, std::vector<std::shared_ptr<Transport>>>>
483 : linkRdmaRes_; // (主链路) 通信域内的RDMA链路，包括通用和专用
484 : std::unordered_map<u32, std::unordered_map<std::string, std::vector<std::shared_ptr<Transport>>>>
485 : linkRdmaResBackUp_; // (备链路) 通信域内的RDMA链路，包括通用和专用
486 :
487 : std::unordered_map<u32, std::unordered_map<std::string, HccltagRemoteResV3>>
488 : rankTagRemoteRes_; // 以rankid&tag粒度保存HccltagRemoteResV3
489 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
490 : usedGeneralLinkNum_; // 记录已经被使用的通信域内通用链路数量
491 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
492 : usedSpecialLinkNum_; // 记录已经被使用的通信域内根据tag构造的链路数量
493 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
494 : usedGeneralLinkRdmaNum_; // 记录已经被使用的通信域内通用RDMA链路数量
495 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
496 : usedSpecialLinkRdmaNum_; // 记录已经被使用的通信域内根据tag构造的RDMA链路数量
497 : std::unordered_map<std::string, AlgResourceResponse> resMap_;
498 :
499 : std::vector<std::vector<std::vector<u32>>> serverAndsuperPodToRank_;
500 : std::vector<std::vector<std::vector<u32>>> commPlaneVector_;
501 : std::vector<bool> isBridgeVector_;
502 : std::shared_ptr<PetersonLock> hostDeviceLock_;
503 : u32 devId_ = 0;
504 : HcclTopoInfo topoInfo_;
505 : HcclAlgoInfo algoInfo_;
506 : std::unique_ptr<TopoMatcher> topoMatcher_;
507 : HcclDispatcher dispatcher_{nullptr};
508 : DeviceMem cclInputBuffer_;
509 : DeviceMem cclOutputBuffer_;
510 : DeviceMem tinySendRecvMem_;
511 : u8 deterministic_ = 0; // 确定性开关
512 : bool dumpDebug_ = false;
513 : bool fftsEnable_ = false; // ffts使能开关
514 : bool inlineReducEnable_ = true; // inline reduce使能
515 : bool interHccsDisable_ = false; // 使能RDMA
516 : u32 multiQpThreshold_{HCCL_MULTI_QP_THRESHOLD_DEFAULT};
517 : u8 debugMode_ = 0; // debug开关
518 : AlgType algType_; // 算法类型
519 : std::string identifier_; // 通信域名称
520 : u64 cclbufferSize_ = 0;
521 : u32 localUserRank_ = 0;
522 : HcclExternalEnable externalEnable_;
523 : std::unordered_map<u32, RankData> rankData_;
524 : std::unordered_map<u32, bool> receivedAcks_;
525 : u64 rpc_;
526 : std::chrono::milliseconds linkTimeOut_; //发送超时时间
527 : // 重执行参数
528 : bool retryEnable_ = false;
529 : u32 retryHoldTime_ = 0;
530 : u32 retryIntervalTime_ = 0;
531 : bool isDeviceMode_ = false; // 区分aicpu和mc2，true表示mc2
532 : u32 mc2OpIndex_ = 0; // mc2算子计数
533 : u32 hcclOpExecIndex_ = 0; // hccl算子执行计数，下沉场景执行计数和下发计数不相等
534 :
535 : std::queue<dfx::ReportStatus> reportStatusQueue_;
536 : std::mutex reportQueueMutex_;
537 : //N秒快恢
538 : bool needsResponseStopLaunch_ = false; //aicpu测试用例下，主线程是否实现停止算子展开
539 : bool isOpLaunch = false; // 算子是否初始化
540 : bool endStopLaunch = false; //主线程/背景线程接收到命令字，是否需要进行处理
541 : bool commOpenStatus = false; //通信域是否可以使用
542 : bool commNeedsRecovery = false; //多通信域下，该通信域是否有故障
543 : bool groupNsCommStatus_ = false; // N秒快恢场景下，流是否被激活
544 : //通用的通道
545 : std::shared_ptr<hccl::HDCommunicate> kfcControlTransferH2D_{nullptr};
546 : std::shared_ptr<hccl::HDCommunicate> kfcStatusTransferD2H_{nullptr};
547 : std::shared_mutex threadAicpuMutex_;
548 : DfxExtendInfo dfxExtendInfo_;
549 : std::vector<SendRecvInfo> allMeshAggregationSendRecvInfo_;
550 : std::shared_ptr<AicpuZeroCopyExchanger> ZeroCopyExchanger_{nullptr};
551 : AicpuHdc aicpuHdc_;
552 : uint64_t groupHashId_{0};
553 : std::map<u32, CqeExceptionStatus> streamCqeExceptionStatus_; // < sqid, status>
554 : HcclSendRecvType bsrRetryOp_{HCCL_SEND_RECV_RESERVED};
555 : HcclOpIdentifier excuteOpId_;
556 : HcclOpIdentifier bsrSendOpId_;
557 : HcclOpIdentifier bsrRecvOpId_;
558 : HcclOpIdentifier bsrTargetOpId_;
559 : u32 bsrSendOpBeginSqePos_ = 0xFFFFFFFF;
560 : u32 bsrRecvOpBeginSqePos_ = 0xFFFFFFFF;
561 : u32 bsrSendOpEndSqePos_ = 0xFFFFFFFF;
562 : u32 bsrRecvOpEndSqePos_ = 0xFFFFFFFF;
563 : u32 bsrSendRetryCnt_ = 0;
564 : u32 bsrRecvRetryCnt_ = 0;
565 : bool bsrSendOpExecException_ = false;
566 : bool bsrRecvOpExecException_ = false;
567 : Stream bsrSendStream_;
568 : Stream bsrRecvStream_;
569 : std::vector<std::vector<HcclSendRecvItem*>> bsrSendRecvPairs_;
570 : // aicpu和custom进程单独对bsr send/recv的index进行计数，用于在重执行过程中保证send/recv的index一致
571 : std::map<u32, u32> bsrSendIndexMap_;
572 : std::map<u32, u32> bsrRecvIndexMap_;
573 : std::map<u32, AicpuStreamMontior> streamTaskMonitor_;
574 :
575 : bool isZeroCopy_{false};
576 : bool isSymmetricMemory_{false};
577 : hccl::AlgOpContext algOpContext_;
578 : std::unique_ptr<HcclTraceInfo> UtraceInfo_;
579 : // taskException
580 : bool printTaskExceptionForErr_ = false; // true表示算子执行异常，需要打印taskException
581 : std::unordered_map<std::string, u32> opTaskException_; // 记录已经打印过taskException的算子信息
582 : // alltoall pipeline
583 : void* sendRecvInfoPtr_ = nullptr;
584 : uint64_t sqeWaitTimeOut_ = dfx::kKfcTimeOut;
585 : uint32_t taskMonitorInterval_ = 0;
586 :
587 : OpCounterInfo opCounterInfo_;
588 : std::mutex queryCqeMutex_;
589 : std::mutex preemptMutexForResMap_;
590 : static bool errMessageReport_;
591 : AicpuKfcHandler kfcHandlers_[static_cast<size_t>(AicpuKfcHandlerType::kMax)]{};
592 :
593 : bool initialized_{ false };
594 :
595 : // 独立算子
596 : bool indOpCommInitialized_{ false }; // 独立算子流程通信域是否初始化
597 : DispatcherCtxPtr dispatcherCtx_{nullptr};
598 : std::unordered_map<std::string, ChannelHandle> channelHandleMap_;
599 : std::unordered_map<ChannelHandle, std::shared_ptr<Transport>> linkMap_;
600 : std::vector<std::shared_ptr<Thread>> threads_;
601 : std::vector<std::unique_ptr<LocalNotify>> notifys_;
602 : TaskException taskExecption_;
603 :
604 : // A3消息语义算子展开aicpu cache
605 : AicpuCacheManager aicpuCacheManager_;
606 :
607 : // 维护aicpu算子展开的索引, 方便定位当前展开的算子信息
608 : size_t opUnfoldIdx_ = 0;
609 : };
610 : } // namespace hccl
611 : #endif // __AICPU_COMMUNICATOR_H__
|
