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
126 : ExecOp(const std::string& newTag, const std::string& algName, OpParam& opParam, const HcclOpResParam* commParam);
127 : HcclResult GetAlgResponseRes(
128 : const std::string& newTag, const std::string& algName, const OpParam& opParam, const HcclOpResParam* commParam,
129 : std::unique_ptr<CollExecutorBase>& executor, AlgResourceResponse*& algResResponse);
130 : void PrepareOpRetryHandler(
131 : u8 inplaceSupportRetry, u8 retryEnable, u8 inPlaceSupportRetryStatus, 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(
153 : const std::string& newTag, const std::string& algName, OpParam& param,
154 : std::unique_ptr<CollExecutorBase>& executor, AlgResourceResponse& algResource, const HcclOpResParam* commParam);
155 : HcclResult SaveTraceInfo(std::string& logInfo);
156 : HcclResult FlushUtraceInfo();
157 : std::string GetExcuteOp();
158 : void HandleCqeException(hccl::Stream& stream, bool isReadClear);
159 : void HandleIndOpCqe();
160 : void ReportIndOpCqe(hccl::Stream& stream, const rtLogicCqReport_t& cqeException, CqeStatus cqeStatus);
161 140 : static void ResetErrMsgReport() { errMessageReport_ = true; };
162 : void PrintTaskExceptionAllComm();
163 : HcclResult PrintTaskExceptionAllThreads();
164 : bool GetOpRetryEnable();
165 : void SetZeroCopyEnable(bool enable);
166 : void SetSymmetricMemoryEnable(bool enable);
167 : bool IsTaskExceptionForHccs();
168 : u32 HcclGetWaitStopExecCmdTimeout();
169 : u32 HcclGetWaitRetryCmdTimeout(uint32_t retryCnt);
170 : HcclResult UpdateOpExecStatus(HcclOpExecFSM& fsmState, KfcStatus state, KfcError& errorCode, uint32_t retryCnt);
171 : HcclResult ResetOpRetryException(HcclCMDType opType);
172 : HcclResult CleanAllRoceResource();
173 : // aclgraph 销毁时通过 KFC 投递清理 aicpu 端单一 tag 的全部关联资源，配合 host 端 AclgraphCallback 使用
174 : HcclResult ClearOpResource(const std::string& tag);
175 : HcclResult SwitchNic();
176 : HcclResult ResumeChangeLink();
177 : HcclResult ParseHierarchicalAlgOption(u32* ahcConfInfo);
178 120 : void RegisterKfcHandler(AicpuKfcHandlerType type, AicpuKfcHandler cb)
179 : {
180 120 : kfcHandlers_[static_cast<size_t>(type)] = cb;
181 120 : }
182 : HcclResult RecordHostOrder(
183 : const HcclOpResParam* commParam, const std::string& tag, u8 orderLaunchMode); // kernel占到核后，通知host侧
184 : std::string GetTaskExceptionTaskInfo(
185 : u32 sqHead, SqeRingBuffer* sqeContextBuffer, uint8_t& type, uint16_t& taskId, uint32_t& remoteRank);
186 : HcclResult RegisterProfCallBack();
187 : // 独立算子专用
188 : HcclResult
189 : SetChannelP2pNotify(TransportDeviceP2pData& transDevP2pData, u64& p2pNotifyNum, HcclChannelP2p& channelP2p);
190 : HcclResult SetChannelRoceNotify(
191 : TransportDeviceIbverbsData& transDevIbverbsData, u64& roceNotifyNum, HcclChannelRoce& channelRoce);
192 : HcclResult InitP2pChannel(HcclIndOpChannelRemoteResV3* commParam, uint32_t channelIndex);
193 : HcclResult InitRoceChannel(HcclIndOpChannelRemoteResV3* commParam, uint32_t channelIndex);
194 : HcclResult AllocChannelResource(HcclIndOpChannelRemoteResV3* commParam);
195 :
196 : HcclResult InitAicpuIndOp(CommAicpuParam* commAicpuParam);
197 0 : bool GetIsInitIndOp() { return indOpCommInitialized_; };
198 : HcclResult InitThreads(ThreadMgrAicpuParam* param);
199 : HcclResult NotifyFree(NotifyMgrAicpuParam* param);
200 : HcclResult NotifyAlloc(NotifyMgrAicpuParam* param);
201 :
202 : HcclResult RegisterOpInfo(void* opInfo, u32 size);
203 : HcclResult RegOpTaskException(HcommGetOpInfoCallback callback);
204 : HcclResult InitProfthreadResource(u32 threadNum);
205 :
206 : HcclResult SetDispatcherCtxOnThread();
207 :
208 : private:
209 : HcclResult SetHrtWorkMode(const HcclOpResParam* commParam);
210 : HcclResult SetHrtDeviceSatMode(const HcclOpResParam* commParam);
211 : HcclResult InitSlaveStreamObjs(const HcclOpResParam* commParam);
212 : HcclResult InitLocalNotifyObj(const HcclOpResParam* commParam);
213 : HcclResult InitOpNotifyObj(const HcclOpResParam* commParam);
214 : HcclResult StreamRestore(u32 streamId); // 将流资源映射到custom进程
215 : HcclResult
216 : ParseTlvToVector(u64 srcTlv, u64 srcTlvTotalLength, std::vector<std::vector<std::vector<u32>>>& vectorInfo);
217 : HcclResult ParseTlvToSubGroupVector(
218 : u64 srcTlv, u64 srcTlvTotalLength, std::vector<std::vector<std::vector<std::vector<u32>>>>& vectorInfo);
219 : HcclResult InitLocalTagRes(const ListCommon& head, bool reAllocFlag = false);
220 : HcclResult InitRemoteTagRes(
221 : u32& rankId, const ListCommon& head, const std::string& newTag, u32 notifyNum,
222 : TransportLinkType linkType = TransportLinkType::RDMA);
223 : template <typename T>
224 : HcclResult InitAndVerifySignal(const HcclSignalInfo& signalInfo, std::vector<std::shared_ptr<T>>& notifyVec);
225 : HcclResult InitTopoMatcher();
226 : HcclResult InitTopoInfo(const HcclOpResParam* commParam);
227 : HcclResult InitCclbuffer(const HcclOpResParam* commParam);
228 : HcclResult InitConfigInfo(const HcclOpResParam* commParam);
229 : HcclResult InitMainStreamObj(const HcclOpResParam* commParam);
230 : HcclResult InitOrderStreamObj(const HcclOpResParam* commParam);
231 : HcclResult InitStreamObj(const HcclStreamParam& streamParam, Stream& stream);
232 : HcclResult InitTimeOutConfig(const HcclOpResParam* commParam);
233 : HcclResult InitHostDeviceLock(const HcclOpResParam* commParam);
234 : HcclResult InitOpRetry(const HcclOpResParam* commParam);
235 : HcclResult InitZeroCopyExchanger(const HcclOpResParam* commParam);
236 : HcclResult
237 : PrepareZeroCopyExchanger(const std::string& newTag, OpParam& opParam, AlgResourceResponse* algResResponse);
238 : HcclResult RegisterDispatcherCallback();
239 : HcclResult RegisterProfilingCallback();
240 : HcclResult InitUtraceInfo(const HcclOpResParam* commParam);
241 : void InitSendRecvOpId(const OpParam& param, HcclOpIdentifier& opId);
242 : HcclResult
243 : GetStreamData(const HcclStreamInfo& streamInfo, HcclComStreamInfo& comStreamInfo, u32& sqHead, u32& sqTail);
244 : HcclResult RefreshTransportsResForRank(
245 : const HcclOpResParam* commParam, u32 rankId, const std::string& newTag, u32 notifyNum,
246 : TransportLinkType linkType = TransportLinkType::RDMA);
247 : HcclResult GetRdmaLinksByRankAndTag(
248 : const HcclOpResParam* commParam, CommTransportsType type, u32 rankId, const std::string& newTag, LINK& link,
249 : bool isBackup, u32 notifyNum, bool isSecond);
250 : HcclResult GetSdmaLinksByRankAndTag(
251 : const HcclOpResParam* commParam, CommTransportsType type, u32 rankId, const std::string& newTag, LINK& link,
252 : bool isBackup, u32 notifyNum, TransportLinkType linkType = TransportLinkType::RESERVED);
253 : HcclResult AllocTransportResource(
254 : const std::string& newTag, const OpParam& opParam, const HcclOpResParam* commParam,
255 : AlgResourceRequest& resRequest, AlgResourceResponse& algResResponse);
256 : HcclResult IncreAllocTransportResource(
257 : const std::string& newTag, const OpParam& opParam, const HcclOpResParam* commParam,
258 : AlgResourceRequest& resRequest, AlgResourceResponse& algResResponse);
259 : HcclResult CreateLink(
260 : const std::string& newTag, TransportRequest& transportRequest, const HcclOpResParam* commParam, LINK& link,
261 : u32 notifyNum, bool isBackup, bool isSecond = false);
262 : HcclResult AllocLocalNotifysResource(
263 : const std::string& newTag, const HcclOpResParam* commParam, const u32 notifyNum,
264 : std::vector<std::shared_ptr<LocalNotify>>& notifiesMain,
265 : std::vector<std::shared_ptr<LocalNotify>>& notifiesAux);
266 : HcclResult AllocStreamsResource(
267 : const std::string& newTag, const HcclOpResParam* commParam, const u32 streamNum, std::vector<Stream>& streams);
268 : HcclResult AllocScratchMemResource(
269 : const std::string& newTag, const HcclOpResParam* commParam, const u64& scratchMemSize, DeviceMem& scratchMem,
270 : bool reAllocFlag = false);
271 : HcclResult AllocAlgResource(
272 : const std::string& newTag, const OpParam& opParam, const HcclOpResParam* commParam,
273 : AlgResourceRequest& resRequest, AlgResourceResponse& algResResponse);
274 : HcclResult CalcResRequest(
275 : const std::string& algName, const OpParam& param, std::unique_ptr<CollExecutorBase>& executor,
276 : AlgResourceRequest& resourceRequest);
277 : HcclResult WaitFinishWhileLoop(
278 : Stream& mainStream, std::vector<Stream>& subStreams, std::string& tag, const uint32_t& beginSqePos,
279 : OpParam& param);
280 : HcclResult CheckOpExecStatusCallback();
281 : HcclResult CheckOpExecStatus();
282 : HcclResult
283 : UpdateSuspendStatus(const OpParam& param, HcclOpExecFSM& fsmState, KfcError& errorCode, uint32_t retryCnt);
284 : HcclResult CheckTaskTimeout(const Stream& mainStream, const uint64_t startUsec);
285 :
286 : HcclResult HcclOpExecFsmInitProcess(
287 : const std::string& newTag, OpParam& param, AlgResourceResponse& algResource, HcclOpExecFSM& fsmState,
288 : KfcError& errorCode);
289 : bool HcclOpCheckSupportRetry(HcclCMDType opType);
290 : HcclResult HcclOpExecChangeLinkProcess(
291 : const std::string& newTag, HcclOpExecFSM& state, KfcError& errorCode, uint32_t& retryCnt,
292 : AlgResourceResponse& algResource, const HcclOpResParam* commParam, const OpParam& param);
293 : HcclResult HcclOpExecFsmLaunchProcess(
294 : const std::string& algName, OpParam& param, std::unique_ptr<CollExecutorBase>& executor,
295 : AlgResourceResponse& algResource, HcclOpExecFSM& fsmState, KfcError& errorCode, uint32_t& beginSqePos,
296 : uint32_t& endSqePos, uint32_t retryCnt);
297 : HcclResult HcclOpExecFsmWaitEndProcess(
298 : OpParam& param, AlgResourceResponse& algResource, HcclOpExecFSM& fsmState, KfcError& errorCode,
299 : uint32_t retryCnt, std::string& tag, const uint32_t& beginSqePos);
300 : HcclResult
301 : HcclOpExecFsmStoppingProcess(const OpParam& param, HcclOpExecFSM& fsmState, KfcError& errorCode, uint32_t retryCnt);
302 : HcclResult HcclOpExecFsmStoppedProcess(
303 : HcclOpExecFSM& fsmState, KfcError& errorCode, uint32_t retryCnt, const std::string& algName, OpParam& param,
304 : uint32_t beginSqePos, uint32_t endSqePos);
305 : HcclResult HcclOpExecFsmWaitRetryProcess(
306 : const OpParam& param, HcclOpExecFSM& fsmState, KfcError& errorCode, KfcCommand& lastCmd);
307 : HcclResult ResetSqBuff();
308 : HcclResult CleanStreamFunc();
309 : HcclResult UpdateSqStatus(Stream& stream);
310 : HcclResult HcclOpExecFsmRetryProcess(
311 : const std::string& algName, OpParam& param, std::unique_ptr<CollExecutorBase>& executor,
312 : AlgResourceResponse& algResource, HcclOpExecFSM& fsmState, KfcError& errorCode, uint32_t& retryCnt,
313 : uint32_t& beginSqePos, uint32_t& endSqePos);
314 : HcclResult RetryOrchestrateHcclOp(
315 : const std::string& algName, OpParam& param, std::unique_ptr<CollExecutorBase>& executor,
316 : AlgResourceResponse& algResource, uint32_t& beginSqePos, uint32_t& endSqePos);
317 : HcclResult HcclOpExecFsmEndProcess(uint32_t retryCnt);
318 : std::string PrintInplaceSupportRetryStatus(InplaceSupportRetryStatus inPlaceSupportRetryStatus);
319 : bool HcclOpSupportRetry(const std::string& algName, bool retryEnable, OpParam& param);
320 : std::string PrintInplaceStatus(u8 isInplaceStatus);
321 : HcclResult SupportRetryWithInplaceCheck(const std::string& algName, OpParam& param);
322 : bool isPollutedZeroCopyOp(OpParam& param);
323 : bool HcclOpCheckNsRecovery();
324 : HcclResult OrchestrateHcclOp(
325 : const std::string& algName, OpParam& param, std::unique_ptr<CollExecutorBase>& executor,
326 : AlgResourceResponse& algResource, uint32_t& beginSqePos, uint32_t& endSqePos);
327 : HcclResult LaunchSlaveStreamTask(AlgResourceResponse& algResource);
328 : HcclResult GetAlltoAllvSendRecvInfo(const void* sendRecvInfoPtr, HcclDataType sendType, HcclDataType recvType);
329 : HcclResult GetAlltoAllvcSendRecvInfo(const void* sendCountMatrix, HcclDataType sendType, HcclDataType recvType);
330 : HcclResult CheckSendRecvParams(const std::vector<SendRecvInfo>& allMeshAggregationSendRecvInfo);
331 : HcclResult SetAlltoAllInputAndOutPutMem(OpParam& param, AlgResourceResponse& algResource);
332 : HcclResult NotifyPost(void);
333 : HcclResult NotifyWait(void);
334 : HcclResult GetAlltoAllTotalCount(OpParam& param, u64& sendCount, u64& recvCount);
335 : HcclResult GetAlltoAllVTotalCount(OpParam& param, u64& sendCount, u64& recvCount);
336 : HcclResult GetAlltoAllVCTotalCount(OpParam& param, u64& sendCount, u64& recvCount);
337 :
338 : // taskException
339 : void
340 : PollCqeException(hccl::Stream& stream, bool isReadClear, rtLogicCqReport_t& cqeException, CqeStatus& cqeStatus);
341 : void ExchangeCqeContext(
342 : hccl::Stream& stream, rtLogicCqReport_t& cqeException, CqeStatus& cqeStatus, ErrCqeContext& cqeCtx);
343 : void ReportErrCqe(hccl::Stream& stream, ErrCqeContext& cqeCtx);
344 : HcclResult PrintTaskExceptionAllStreams();
345 : bool IsRepeatedOpTaskException(u32 idx, SqeRingBuffer* sqeContextBuffer); // 避免同一个算子重复打印taskException
346 : std::string GetTaskExceptionOpInfo(u32 idx, SqeRingBuffer* sqeContextBuffer); // 打印算子参数信息
347 : void PrintTaskExceptionTaskQue(
348 : u32 sqIdx, SqeRingBuffer* sqeContextBuffer, bool isMonitor = false); // 打印当前位置的前序task
349 : std::string GetTaskBriefsInfo(u32 idx, SqeRingBuffer* sqeContextBuffer); // 打印task简写
350 : void PrintAicpuCommExecStatus();
351 :
352 : HcclResult UpdateOpRingBufferIdx();
353 : HcclResult CombineReportOpInfo(OpParam& param, bool isRetry, bool isRelay);
354 : void UpdateBSRRetryCnt();
355 : void ResetBSRRetryCnt();
356 : HcclResult CommitBSRStoredException(HcclOpExecFSM& fsmState, KfcError& errorCode);
357 : HcclResult QueryBatchSendRecvPairBeginPos();
358 : HcclResult QueryBatchSendRecvPairEndPos();
359 : HcclResult UpdateOpExecStatus(
360 : HcclOpExecFSM& fsmState, HcclOpIdentifier& opId, KfcStatus state, KfcError& errorCode, uint32_t retryCnt);
361 : u32 HcclUpdateBatchSendRecvOpIndex(std::map<u32, u32>& bsrIndexMap, u32 peerRank);
362 : u32 HcclUpdateBatchSendRecvOpIndex(HcclSendRecvType opType, u32 srcRank, u32 dstRank);
363 : HcclResult InitBatchSendRecvOpId(
364 : const OpParam& param, const HcclSendRecvItem* sendrecvPair, HcclOpIdentifier& opId, u32 streamId,
365 : AlgResourceResponse& algResource);
366 : HcclResult InitBatchSendRecvOpId(const OpParam& param, AlgResourceResponse& algResource);
367 : HcclResult InitBsrSendRecvOpIdAndExcuteOpId(
368 : OpParam& param, AlgResourceResponse& algResource, HcclOpExecFSM& fsmState, KfcError& errorCode);
369 : void SetBSRSendOpExecException();
370 : void SetBSRRecvOpExecException();
371 : bool GetBSRSendOpExecException();
372 : bool GetBSRRecvOpExecException();
373 :
374 : HcclResult CleanStream(Stream& stream);
375 : HcclResult ClearStreamCqeException(Stream& stream);
376 : HcclResult ResetBSRSendOpExecException();
377 : HcclResult ResetBSRRecvOpExecException();
378 : HcclResult ResetBSRException();
379 : HcclResult BSRStopedProcess(HcclOpExecFSM& fsmState, KfcError& errorCode);
380 : HcclResult GetBSRRetryOpId(const OpParam& param, HcclOpIdentifier& targetOpId);
381 : HcclResult InitExecLoop(OpParam& param, std::unique_ptr<CollExecutorBase>& executor, u32& loopNum);
382 : template <typename T>
383 : HcclResult InitAndVerifySingleSignal(const HcclSignalInfo& signalInfo, std::shared_ptr<T>& notify);
384 : HcclResult SetTransportMachinePara(
385 : MachinePara& machinePara, u32& rankId, const std::string& newTag,
386 : TransportLinkType linkType = TransportLinkType::RESERVED);
387 : HcclResult CheckNotifyOrQPMaxNum(u64& existNum, const u64& MaxNum, const bool& isNotifyRes);
388 : HcclResult SetTagRemoteRes(u32& rankId, const std::string& tag, HccltagRemoteResV2* tagRes);
389 : HcclResult SetTransportPtpNotify(
390 : TransportDeviceP2pData& transDevP2pData, u64& p2pNotifyNum, HcclLinkP2pV2& linkP2p, u32 notifyNum);
391 : HcclResult
392 : SetTransportRoceQP(TransportDeviceIbverbsData& transDevIbverbsData, u64& roceQpNum, HcclLinkRoceV2* linkRoce);
393 : HcclResult SetTransportRoceNotify(
394 : TransportDeviceIbverbsData& transDevIbverbsData, u64& roceNotifyNum, HcclLinkRoceV2* linkRoce, u32 notifyNum);
395 : HcclResult InitLinkP2p(
396 : HccltagRemoteResV2* tagRes, u32& rankId, const std::string& newTag, u32 notifyNum,
397 : TransportLinkType linkType = TransportLinkType::RESERVED);
398 : HcclResult InitLinkRoce(
399 : HccltagRemoteResV2* tagRes, u32& rankId, const std::string& newTag, u32 notifyNum, const bool isBackup = false);
400 : HcclResult InitLinkRoce(
401 : HccltagRemoteResV2* tagRes, HcclLinkRoceV2* linkRoce, u32& rankId, const std::string& newTag, u32 notifyNum,
402 : const bool isBackup = false, const bool isSecond = false);
403 : HcclResult GetBsrTransportQpn(const HcclSendRecvItem* sendrecvPair, AlgResourceResponse& algResource, u32& qpn);
404 : HcclResult ReAllocTransportResource(
405 : const std::string& newTag, AlgResourceResponse& algResResponse, std::map<u32, bool>& remoteRankPortMap,
406 : const HcclOpResParam* commParam, const OpParam& param);
407 : HcclResult CleanRoceResource(
408 : const std::string& newTag, AlgResourceResponse& algResResponse, const std::map<u32, bool>& remoteRankPortMap,
409 : const OpParam& param);
410 : HcclResult LoadChangeLinkInfo(ChangeLinkInfo& changeLinkInfo);
411 :
412 : HcclResult AddRetryExecFlipTask(AlgResourceResponse& algResource);
413 : HcclResult ReportHcclTaskInfo(Stream& mainStream, std::vector<Stream>& subStreams);
414 : HcclResult ClearLocalBuff(Stream& mainStream, std::vector<Stream>& subStreams);
415 : HcclResult UpdateProfReportStartSqeIdx();
416 : HcclResult TasktypeTransferD2H(const uint8_t sqeType, TaskType& taskType);
417 : void PrepareMc2Handler();
418 : HcclResult InitOpCounter(const OpCounterInfo& opCounterInfo);
419 :
420 : // rts调用接口,通过mailbox上报给tsfw
421 : HcclResult SendTaskExceptionByMBox(const uint16_t& rsErrorCode);
422 :
423 : HcclResult RefreshLinkForSwitchNic(
424 : const std::string& newTag, const TransportRequest& transportRequest,
425 : const std::map<u32, bool>& remoteRankPortMap, bool isSecondBuild, LINK& switchLink);
426 : HcclResult ReAllocTransportForSwitchNic(
427 : const std::string& newTag, AlgResourceResponse& algResResponse, std::map<u32, bool>& remoteRankPortMap);
428 : HcclResult RefreshRoceTransportsForSwitchNic(std::unordered_map<std::string, OpCommTransport>& reservedLinks);
429 : HcclResult RevertTransportsForSwitchNic(std::unordered_map<std::string, OpCommTransport>& reservedLinks);
430 : HcclResult SwitchNicWaitHandleCommand(std::unordered_map<std::string, OpCommTransport>& reservedLinks);
431 : HcclResult SwitchNicWaitResult(std::unordered_map<std::string, OpCommTransport>& reservedLinks);
432 : u32 CalculateOpExecIndex(const OpParam& opParam, u32 userRank); // 每次展开时计算
433 :
434 : HcclResult InitProfResource();
435 : void InitCommInfoStatus(bool commInfo);
436 : HcclResult InitTinyMem(const HcclOpResParam* commParam);
437 : HcclResult SetStreamEnable(Stream& stream);
438 : HcclResult RefreshAlgResponseTransportRes(
439 : const std::string& newTag, AlgResourceResponse& algResResponse, std::map<u32, bool>& remoteRankPortMap,
440 : bool isChangeLinkFlag, const HcclOpResParam* commParam, const OpParam& param);
441 : HcclResult RefreshCommResponseTransportRes(std::map<u32, bool>& remoteRankPortMap);
442 : HcclResult PrintTaskExceptionByTaskId(u8 sqeType, u16 taskId, hccl::Stream& stream, u32 tail);
443 : bool IsNoNeedWait(void);
444 : void SetStreamCqeExceptionStatus(const Stream& stream, CqeExceptionStatus cqeStatus);
445 : void ResetStreamCqeExceptionStatus(const Stream& stream);
446 : CqeExceptionStatus GetStreamCqeExceptionStatus(const Stream& stream);
447 : HcclResult GenTaskExceptionInfo(u8 sqeType, hccl::Stream& stream, u32 head);
448 : HcclResult InvokeKfcHandler(AicpuKfcHandlerType type, const std::vector<u64> args);
449 :
450 : bool IsNoNeedMonitor(void);
451 : void InsertMonitorData(Stream& stream, HcclUs& curTime, u32 sqHead, uint16_t taskId, uint8_t type);
452 : bool IsNeedRefreshMonitorData(
453 : AicpuStreamMontior& streamMontior, HcclUs& curTime, uint32_t remoteRank, uint16_t taskId, u32 sqHead,
454 : u32 sqTail, uint8_t type);
455 : // 对称内存
456 : HcclResult PrepareSymmetricMemory(const OpParam& param, OpCommTransport& opTransportResponse);
457 : HcclResult PrepareSymmetricMemRanges(
458 : const AlgResourceResponse& algResource, uint64_t inputSize, uint64_t outputSize,
459 : std::vector<OpUnfoldMemRange>& userInputMemRanges, std::vector<OpUnfoldMemRange>& userOutputMemRanges);
460 :
461 : HcclResult CalSendRecvInfoForAlltoall(const OpParam& param);
462 : HcclResult CalSendRecvInfoFor910B(
463 : const std::string& algName, const OpParam& param, std::unique_ptr<CollExecutorBase>& executor);
464 : void HandleExistTagReAlloc(
465 : HccltagLocalResV2* tagRes, const std::string& tag, bool reAllocFlag, ListCommon*& curList, bool& needSkip);
466 :
467 : // 获取指定索引的算子信息
468 : const AicpuOpInfo* GetOpInfoFromSqIdx(u32 sqIdx, SqeRingBuffer* sqeContextBuffer);
469 :
470 : // 打印算子数据信息
471 : void PrintOpDataInfo(u32 sqIdx, SqeRingBuffer* sqeContextBuffer, bool isMonitor);
472 :
473 : // 打印task序列行
474 : void PrintTaskLine(bool isMonitor, u32 lineNum, u32 totalPrinted, const std::string& taskLine) const;
475 :
476 : // 更新算子上下文
477 : void UpdateOpContext(
478 : u32& opIndex, std::string& opTag, u32& lineCount, std::vector<std::string>& currentOpTasks, u32 newOpIndex,
479 : const std::string& newOpTag) const;
480 :
481 : // 准备下一行数据
482 : void PrepareNextLine(u32 opIndex, u32& lineCount, std::vector<std::string>& currentOpTasks) const;
483 :
484 : // 拼接task列表为字符串
485 : std::string ConcatTaskLine(const std::vector<std::string>& tasks) const;
486 :
487 : // 打印剩余未满行的tasks
488 : void PrintRemainingTasks(
489 : bool isMonitor, u32 lineCount, u32 printedCount, const std::vector<std::string>& currentOpTasks) const;
490 :
491 : std::unordered_map<s32, u32> opExecIndexMap_;
492 :
493 : // 管理aicpu和custom进程共享的数据
494 : AicpuShareDataManager aicpuShareData_;
495 : bool isCustom_ = false;
496 :
497 : // local资源
498 : std::vector<Stream> slaveStreams_;
499 : Stream mainStream_;
500 : Stream orderStream_;
501 : std::unordered_set<u32> streamToObj_; // 从context的资源构造为Stream对象去重
502 : s32 userStreamId_; // 用户传入的stream的id
503 :
504 : std::vector<std::shared_ptr<LocalNotify>> localNotifies_; // 主从流之间同步的notify
505 : std::vector<std::shared_ptr<LocalNotify>> opNotifies_; // host与device间同步的notify
506 : std::vector<std::shared_ptr<LocalNotify>> orderNotifies_{AICPU_ORDER_NOTIFY_MAX_NUM, nullptr}; // 按序下发的notify
507 : std::unordered_set<u32> notifysToObj_; // 从context的资源构造为LocalNotify对象去重
508 :
509 : std::unordered_map<std::string, std::shared_ptr<DeviceMem>> tagScratchMem_; // 本地scratchmem
510 : std::unordered_map<std::string, std::unordered_set<u64>> localTagResToObj_; // 从context的localtag资源构造为对象去重
511 :
512 : // 跨卡资源
513 : uint32_t notifySize_;
514 : const HcclOpResParam* commParam_ = nullptr;
515 :
516 : // 通信域内的link
517 : std::unordered_map<u32, std::unordered_map<std::string, std::shared_ptr<Transport>>>
518 : linkRes_; // 通信域内的SDMA hccs链路，包括通用和专用
519 : // 通信域内的SDMA sio链路
520 : std::unordered_map<u32, std::unordered_map<std::string, std::shared_ptr<Transport>>> linkResSio_;
521 : std::unordered_map<u32, std::unordered_map<std::string, std::vector<std::shared_ptr<Transport>>>>
522 : linkRdmaRes_; // (主链路) 通信域内的RDMA链路，包括通用和专用
523 : std::unordered_map<u32, std::unordered_map<std::string, std::vector<std::shared_ptr<Transport>>>>
524 : linkRdmaResBackUp_; // (备链路) 通信域内的RDMA链路，包括通用和专用
525 :
526 : std::unordered_map<u32, std::unordered_map<std::string, HccltagRemoteResV3>>
527 : rankTagRemoteRes_; // 以rankid&tag粒度保存HccltagRemoteResV3
528 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
529 : usedGeneralLinkNum_; // 记录已经被使用的通信域内通用链路数量
530 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
531 : usedSpecialLinkNum_; // 记录已经被使用的通信域内根据tag构造的链路数量
532 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
533 : usedGeneralLinkRdmaNum_; // 记录已经被使用的通信域内通用RDMA链路数量
534 : std::unordered_map<u32, std::unordered_map<std::string, u32>>
535 : usedSpecialLinkRdmaNum_; // 记录已经被使用的通信域内根据tag构造的RDMA链路数量
536 : std::unordered_map<std::string, AlgResourceResponse> resMap_;
537 :
538 : std::vector<std::vector<std::vector<u32>>> serverAndsuperPodToRank_;
539 : std::vector<std::vector<std::vector<u32>>> commPlaneVector_;
540 : std::vector<bool> isBridgeVector_;
541 : std::shared_ptr<PetersonLock> hostDeviceLock_;
542 : u32 devId_ = 0;
543 : HcclTopoInfo topoInfo_;
544 : HcclAlgoInfo algoInfo_;
545 : std::unique_ptr<TopoMatcher> topoMatcher_;
546 : HcclDispatcher dispatcher_{nullptr};
547 : DeviceMem cclInputBuffer_;
548 : DeviceMem cclOutputBuffer_;
549 : DeviceMem tinySendRecvMem_;
550 : u8 deterministic_ = 0; // 确定性开关
551 : bool dumpDebug_ = false;
552 : bool fftsEnable_ = false; // ffts使能开关
553 : bool inlineReducEnable_ = true; // inline reduce使能
554 : bool interHccsDisable_ = false; // 使能RDMA
555 : u32 multiQpThreshold_{HCCL_MULTI_QP_THRESHOLD_DEFAULT};
556 : u8 debugMode_ = 0; // debug开关
557 : AlgType algType_; // 算法类型
558 : std::string identifier_; // 通信域名称
559 : u64 cclbufferSize_ = 0;
560 : u32 localUserRank_ = 0;
561 : HcclExternalEnable externalEnable_;
562 : std::unordered_map<u32, RankData> rankData_;
563 : std::unordered_map<u32, bool> receivedAcks_;
564 : u64 rpc_;
565 : std::chrono::milliseconds linkTimeOut_; // 发送超时时间
566 : // 重执行参数
567 : bool retryEnable_ = false;
568 : u32 retryHoldTime_ = 0;
569 : u32 retryIntervalTime_ = 0;
570 : bool isDeviceMode_ = false; // 区分aicpu和mc2，true表示mc2
571 : u32 mc2OpIndex_ = 0; // mc2算子计数
572 : u32 hcclOpExecIndex_ = 0; // hccl算子执行计数，下沉场景执行计数和下发计数不相等
573 :
574 : std::queue<dfx::ReportStatus> reportStatusQueue_;
575 : std::mutex reportQueueMutex_;
576 : // N秒快恢
577 : bool needsResponseStopLaunch_ = false; // aicpu测试用例下，主线程是否实现停止算子展开
578 : bool isOpLaunch = false; // 算子是否初始化
579 : bool endStopLaunch = false; // 主线程/背景线程接收到命令字，是否需要进行处理
580 : bool commOpenStatus = false; // 通信域是否可以使用
581 : bool commNeedsRecovery = false; // 多通信域下，该通信域是否有故障
582 : bool groupNsCommStatus_ = false; // N秒快恢场景下，流是否被激活
583 : // 通用的通道
584 : std::shared_ptr<hccl::HDCommunicate> kfcControlTransferH2D_{nullptr};
585 : std::shared_ptr<hccl::HDCommunicate> kfcStatusTransferD2H_{nullptr};
586 : std::shared_mutex threadAicpuMutex_;
587 : DfxExtendInfo dfxExtendInfo_;
588 : std::vector<SendRecvInfo> allMeshAggregationSendRecvInfo_;
589 : std::shared_ptr<AicpuZeroCopyExchanger> ZeroCopyExchanger_{nullptr};
590 : AicpuHdc aicpuHdc_;
591 : uint64_t groupHashId_{0};
592 : std::map<u32, CqeExceptionStatus> streamCqeExceptionStatus_; // < sqid, status>
593 : HcclSendRecvType bsrRetryOp_{HCCL_SEND_RECV_RESERVED};
594 : HcclOpIdentifier excuteOpId_;
595 : HcclOpIdentifier bsrSendOpId_;
596 : HcclOpIdentifier bsrRecvOpId_;
597 : HcclOpIdentifier bsrTargetOpId_;
598 : u32 bsrSendOpBeginSqePos_ = 0xFFFFFFFF;
599 : u32 bsrRecvOpBeginSqePos_ = 0xFFFFFFFF;
600 : u32 bsrSendOpEndSqePos_ = 0xFFFFFFFF;
601 : u32 bsrRecvOpEndSqePos_ = 0xFFFFFFFF;
602 : u32 bsrSendRetryCnt_ = 0;
603 : u32 bsrRecvRetryCnt_ = 0;
604 : bool bsrSendOpExecException_ = false;
605 : bool bsrRecvOpExecException_ = false;
606 : Stream bsrSendStream_;
607 : Stream bsrRecvStream_;
608 : std::vector<std::vector<HcclSendRecvItem*>> bsrSendRecvPairs_;
609 : // aicpu和custom进程单独对bsr send/recv的index进行计数，用于在重执行过程中保证send/recv的index一致
610 : std::map<u32, u32> bsrSendIndexMap_;
611 : std::map<u32, u32> bsrRecvIndexMap_;
612 : std::map<u32, AicpuStreamMontior> streamTaskMonitor_;
613 :
614 : bool isZeroCopy_{false};
615 : bool isSymmetricMemory_{false};
616 : hccl::AlgOpContext algOpContext_;
617 : std::unique_ptr<HcclTraceInfo> UtraceInfo_;
618 : // taskException
619 : bool printTaskExceptionForErr_ = false; // true表示算子执行异常，需要打印taskException
620 : std::unordered_map<std::string, u32> opTaskException_; // 记录已经打印过taskException的算子信息
621 : // alltoall pipeline
622 : void* sendRecvInfoPtr_ = nullptr;
623 : uint64_t sqeWaitTimeOut_ = dfx::kKfcTimeOut;
624 : uint32_t taskMonitorInterval_ = 0;
625 :
626 : OpCounterInfo opCounterInfo_;
627 : std::mutex queryCqeMutex_;
628 : std::mutex preemptMutexForResMap_;
629 : static bool errMessageReport_;
630 : AicpuKfcHandler kfcHandlers_[static_cast<size_t>(AicpuKfcHandlerType::kMax)]{};
631 :
632 : bool initialized_{false};
633 :
634 : // 独立算子
635 : bool indOpCommInitialized_{false}; // 独立算子流程通信域是否初始化
636 : DispatcherCtxPtr dispatcherCtx_{nullptr};
637 : std::unordered_map<std::string, ChannelHandle> channelHandleMap_;
638 : std::unordered_map<ChannelHandle, std::shared_ptr<Transport>> linkMap_;
639 : std::vector<std::shared_ptr<Thread>> threads_;
640 : std::vector<std::unique_ptr<LocalNotify>> notifys_;
641 : TaskException taskExecption_;
642 :
643 : // A3消息语义算子展开aicpu cache
644 : AicpuCacheManager aicpuCacheManager_;
645 :
646 : // 维护aicpu算子展开的索引, 方便定位当前展开的算子信息
647 : size_t opUnfoldIdx_ = 0;
648 : };
649 : } // namespace hccl
650 : #endif // __AICPU_COMMUNICATOR_H__
|
