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 : #ifndef HCCL_COMMUNICATOR_IMPL_H
11 : #define HCCL_COMMUNICATOR_IMPL_H
12 :
13 : #include <set>
14 : #include <vector>
15 : #include <unordered_map>
16 : #include <string>
17 : #include <atomic>
18 : #include "types.h"
19 : #include "coll_service_default_impl.h"
20 : #include "coll_service_ai_cpu_impl.h"
21 : #include "conn_local_notify_manager.h"
22 : #include "conn_local_cnt_notify_manager.h"
23 : #include "data_buf_manager.h"
24 : #include "local_rma_buf_manager.h"
25 : #include "queue_notify_manager.h"
26 : #include "remote_rma_buf_manager.h"
27 : #include "rma_conn_manager.h"
28 : #include "stream_manager.h"
29 : #include "queue_wait_group_cnt_notify_manager.h"
30 : #include "host_device_sync_notify_manager.h"
31 : #include "queue_bcast_post_cnt_notify_manager.h"
32 : #include "mem_transport_manager.h"
33 : #include "rank_gph.h"
34 : #include "rank_graph_builder.h"
35 : #include "snap_shot_parse.h"
36 : #include "comm_type.h"
37 : #include "mirror_task_manager.h"
38 : #include "aicpu_stream_manager.h"
39 : #include "profiling_reporter.h"
40 : #include "hdc.h"
41 : #include "hccl_one_sided_service.h"
42 : #include "coll_alg_component.h"
43 : #include "ub_memory_transport_mgr.h"
44 : #include "ccu_super_fast_load.h"
45 : #include "ccu_stream_sync_notify_manager.h"
46 : #include "ccu_driver_handle.h"
47 : #include "hccl_common_v2.h"
48 : #include "hccl_rank_graph.h"
49 : #include "hccl_aiv_utils.h"
50 : #include "error_message_v2.h"
51 : #include "hccp.h"
52 : #include "aicpu/launch_device.h"
53 : #include "communicator_dpu_types.h"
54 :
55 : namespace Hccl {
56 :
57 : using HcclUs = std::chrono::steady_clock::time_point;
58 : class CommunicatorImpl {
59 : public:
60 : HcclResult Init(const CommParams &commParams, const std::string &rankTablePath);
61 : HcclResult Init(const CommParams &commParams, const std::string &ranktableM, const HcclCommConfig &config);
62 : HcclResult Init(const CommParams &commParams, const RankTableInfo &ranktable, const HcclCommConfig &config);
63 :
64 : HcclResult CreateSubComm(const CommParams &subCommParams, const std::vector<u32> &rankIds,
65 : CommunicatorImpl *subCommImpl);
66 : HcclResult CreateSubComm(const CommParams &subCommParams, const std::vector<u32> &rankIds,
67 : CommunicatorImpl *subCommImpl, HcclCommConfig &subConfig);
68 :
69 : HcclResult LoadOpbasedCollOp(const CollOpParams &opParams, void *stream);
70 :
71 : HcclResult AllocCollOpResource(const CollOpParams &opParams, void **addr);
72 :
73 : HcclResult AllocCommResource(void *mc2Tiling, void **commContext);
74 :
75 : HcclResult GetCcuTaskInfo(void *tilingData, void *ccuTaskGroup) const;
76 : HcclResult GetTopoDesc(HcclTopoDescs *topoDescs, uint32_t topoSize) const;
77 :
78 : HcclResult CalcCollOffloadOpRes(const OpType opType, u64 dataSize, HcclDataType dataType, CollOffloadOpResReq &resReq);
79 : HcclResult SetCollOffloadSlaveStreams(const std::string &opTag, std::vector<void *> slaveStreams);
80 : HcclResult SetCollOffloadScratchBuf(const std::string &opTag, void *scratchMemPtr, u64 requiredScratchMemSize);
81 : HcclResult LoadOffloadCollOp(std::string &opTag, const CollOpParams &opParams, void *stream);
82 :
83 : HcclResult SaveTopoDesc(std::string &identifier);
84 : HcclResult GetConfigInCCLbufferSize(uint64_t *cclBufSize);
85 : HcclResult GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum);
86 : HcclResult GetInstSizeByNetLayer(uint32_t netLayer, uint32_t *rankNum);
87 : HcclResult GetInstTopoTypeByNetLayer(uint32_t netLayer, uint32_t *topoType);
88 : HcclResult GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **ranks, uint32_t *rankNum);
89 : HcclResult GetInstSizeListByNetLayer(uint32_t netLayer, uint32_t **instSizeList, uint32_t *listSize);
90 : HcclResult GetLinks(uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink **linkList, uint32_t *listSize);
91 : HcclResult GetTopoInstsByLayer(uint32_t netLayer, uint32_t **topoInsts, uint32_t *topoInstNum);
92 : HcclResult GetTopoType(uint32_t netLayer, uint32_t topoInstId, CommTopo *topoType);
93 : HcclResult GetRanksByTopoInst(uint32_t netLayer, uint32_t topoInstId, uint32_t **ranks, uint32_t *rankNum);
94 :
95 : HcclResult GetEndpointNum(uint32_t layer, uint32_t topoInstId, uint32_t* num);
96 : HcclResult GetEndpointDesc(uint32_t layer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc);
97 : HcclResult GetEndpointInfo(uint32_t rankId, const EndpointDesc *endPointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void *info);
98 :
99 : u32 GetCcuMc2ServerNum();
100 :
101 : const string &GetId() const;
102 :
103 : u32 GetIdIndex() const;
104 :
105 : RankId GetMyRank() const;
106 :
107 : u32 GetRankSize() const;
108 :
109 : u32 GetDeviceLogicId() const;
110 :
111 : u32 GetDevicePhyId() const;
112 :
113 : u64 GetBufferSize() const;
114 :
115 : const DevType &GetDevType() const;
116 :
117 : shared_ptr<RankGraph> GetRankGraph() const;
118 :
119 : bool GetOpAiCpuTSFeatureFlag() const;
120 :
121 : bool GetOpAivFeatureFlag() const;
122 :
123 : bool GetOpCcuFeatureFlag() const;
124 :
125 : bool GetCommAiCpuTSFeatureFlag() const;
126 :
127 : bool GetCommCcuFeatureFlag() const;
128 :
129 : virtual DataBufManager &GetDataBufferManager() const; // NOTE:添加 virtual用于UT打桩
130 :
131 : virtual LocalRmaBufManager &GetLocalRmaBufManager() const;
132 :
133 : virtual RemoteRmaBufManager &GetRemoteRmaBufManager() const;
134 :
135 : virtual QueueNotifyManager &GetAicpuQueueNotifyManager() const;
136 :
137 : virtual QueueNotifyManager &GetCcuQueueNotifyManager() const;
138 :
139 : virtual ConnLocalNotifyManager &GetConnLocalNotifyManager() const;
140 :
141 : virtual ConnLocalCntNotifyManager &GetConnLocalCntNotifyManager() const;
142 :
143 : virtual QueueWaitGroupCntNotifyManager &GetQueueWaitGroupCntNotifyManager() const;
144 :
145 : virtual QueueBcastPostCntNotifyManager &GetBcastPostCntNotifyManager() const;
146 :
147 : virtual StreamManager &GetStreamManager() const;
148 :
149 : virtual AicpuStreamManager &GetAicpuStreamManager() const;
150 :
151 : virtual CollServiceBase *GetCollService() const;
152 :
153 : virtual CollServiceBase *GetCcuCollService() const;
154 :
155 : virtual SocketManager &GetSocketManager() const;
156 :
157 : virtual RmaConnManager &GetRmaConnManager() const;
158 :
159 : virtual const string &GetEstablishLinkSocketTag() const;
160 :
161 : virtual CollOperator *GetCurrentCollOperator() const;
162 :
163 : virtual MemTransportManager *GetMemTransportManager() const;
164 :
165 : virtual HostDeviceSyncNotifyManager &GetHostDeviceSyncNotifyManager() const;
166 :
167 : virtual Trace &GetTrace() const;
168 :
169 : virtual u32 GetOpBaseOpIndex() const;
170 :
171 : virtual u32 GetOpIndex() const;
172 :
173 : u32 GetSubmittedOpCnt() const;
174 :
175 : HDCommunicate &GetKfcControlTransferH2D() const;
176 :
177 : HDCommunicate &GetKfcStatusTransferD2H() const;
178 :
179 : HcclResult Suspend();
180 :
181 : HcclResult Clean();
182 :
183 : HcclResult Resume();
184 :
185 5 : void SetAicpuKernelLaunched(bool flag)
186 : {
187 5 : isAicpuKernelLaunched = flag;
188 5 : }
189 :
190 : const NotifyTimeoutCfg &GetNotifyTimeoutCfg() const;
191 :
192 283 : const shared_ptr<DevBuffer> &GetCclBuffer() const
193 : {
194 : // 接口设计不合理，其他数据结构无权管理hccl buffer生命周期
195 : // 建议改为传递裸指针，但整改范围较大，本次性能优化暂不处理
196 283 : return cclBuffer;
197 : }
198 :
199 1 : const shared_ptr<DevBuffer> GetAivTagBuffer() const
200 : {
201 1 : return aivTagBuffer;
202 : }
203 :
204 8 : const shared_ptr<DevBuffer> GetAivOffloadTagBuffer() const
205 : {
206 8 : return aivOffloadTagBuffer;
207 : }
208 :
209 : const shared_ptr<DevBuffer> GetInCclBuffer() const
210 : {
211 : return inCclBuffer;
212 : }
213 :
214 : const shared_ptr<DevBuffer> GetOutCclBuffer() const
215 : {
216 : return outCclBuffer;
217 : }
218 :
219 0 : const shared_ptr<DevBuffer> GetKFCWorkSpace(const char *memTag) const
220 : {
221 0 : std::string tag = memTag != nullptr ? std::string(memTag) : "";
222 0 : auto it = tagWorkspaceMap_.find(tag);
223 0 : return it != tagWorkspaceMap_.end() ? it->second : nullptr;
224 0 : }
225 :
226 : HcclResult CreateCommCclBuf();
227 : HcclResult GetInCclBuf(void* &commInputPtr, u64 &commInputSize);
228 : HcclResult GetOutCclBuf(void* &commOutputPtr, u64 &commOutputSize);
229 : HcclResult GetIndirectInCclBuf(void* &commIndirectInputPtr, u64 &commIndirectInputSize);
230 : HcclResult GetIndirectOutCclBuf(void* &commIndirectOutputPtr, u64 &commIndirectOutputSize);
231 :
232 : HcclResult GetLocalCclBuffer(void **addr, uint64_t *size);
233 : HcclResult GetDevMemWorkSpace(const std::string &memTag, uint64_t *size, void **addr, bool *newCreated);
234 : HcclResult CreateWorkspaceBuf(const char *memTag, uint64_t *size, bool *newCreated);
235 : HcclResult AllocAndRegKFCWorkSpace(uint64_t size, const std::string &memTag);
236 : HcclResult CleanupKFCWorkSpaceOnFailure(DpuShmem &shmem, HcclResult ret);
237 : HcclResult GetKFCWorkSpaceVA(const std::string &memTag, const uint64_t *size, void **addr, bool *newCreated);
238 : HcclResult DestroyKFCWorkSpaceVA();
239 :
240 : bool IsWorldGroup() const;
241 :
242 : // 静态信息序列化,供获取保存快照size时使用
243 5 : BinaryStream &GetStaticBinaryInfo()
244 : {
245 5 : return staticBinaryInfo;
246 : }
247 : // 获取rankTable的字节流，供一致性校验crc时带localId使用
248 : u32 GetRanktableCrc(bool isContainLoaId) const;
249 : HcclResult RecoverRankGraphData(SnapShotComm &snapShotComm, const char *changeInfo);
250 : HcclResult RecoverTransportData(u32 savedSubmittedOpCnt, const std::vector< std::pair<u32, RankId>> &levelRankPairs, u32 savedStep, vector<std::pair<LinkGroup, u32>> linkGroupPair);
251 : HcclResult RecoverExeCfgData(const OpExecuteConfig& inOpExeCfg, const OpExecuteConfig &inCommExeCfg, bool inIsLoadOp);
252 : virtual HcclResult GetSnapShotDynamicBuf(BinaryStream &buf) const;
253 :
254 : HcclResult RecoverComm(SnapShotComm &snapShotComm, u32 stepParam, const char *changeInfo);
255 : HcclResult RecoverComm(const SnapShotSubComm &snapShotSubComm, std::unique_ptr<RankGraph> &inputRankGraph, u32 inputStep);
256 : HcclResult RecoverSubComm(const SnapShotSubComm &snapShotSubComm, CommunicatorImpl *subCommImpl, u32 step);
257 : HcclResult RecoverOpMode(u32 opMode);
258 : std::set<RankId> GetNeighboorRanks() const;
259 : virtual u32 GetCollOpIndex() const;
260 :
261 : virtual u32 GetStep() const;
262 : bool IsCommReady();
263 : void CovertToCurrentCollOperator(std::string &opTag, const CollOpParams &opParams, OpMode opMode, bool isLaunch = true, bool isHcomSelectAlg = false);
264 :
265 : virtual MirrorTaskManager &GetMirrorTaskManager() const;
266 : virtual ProfilingReporter &GetProfilingReporter() const;
267 :
268 : ~CommunicatorImpl();
269 : HcclResult NotifyAicpuDestroyComm();
270 :
271 : virtual HcclResult GetOneSidedService(HcclOneSidedService** service) const;
272 : u32 GetUsedChannelCount(u32 dieId);
273 : void PrintChannelInfoCallback() const;
274 : void RegisterPrintChannelInfoCallback(std::function<void()> callback);
275 : void SetCommStatus(CommStatus commStatus);
276 : CommStatus GetCommStatus() const;
277 :
278 : /* mc2数据上报 */
279 : void ReportHcclMC2Info(const Stream &kfcStream, Stream &stream, const std::vector<Stream*> &aicpuStreams);
280 :
281 464 : const OpExecuteConfig& GetOpExecuteConfig() const // 获取算子粒度 加速模式
282 : {
283 464 : return opExecuteConfig;
284 : }
285 137 : const OpExecuteConfig& GetCommExecuteConfig() const // 获取通讯域粒度 加速模式
286 : {
287 137 : return commExecuteConfig;
288 : }
289 : void SetOpExecuteConfig(const OpExecuteConfig& inConfig);
290 : void SetCommExecuteConfig(const OpExecuteConfig& inConfig);
291 43 : const std::string& GetCurAlgName() const
292 : {
293 43 : return curAlgName;
294 : }
295 31 : CollAlgComponent* GetCollAlgComponent()
296 : {
297 31 : return collAlgComponent.get();
298 : }
299 :
300 : std::map<AivOpCacheArgs, std::shared_ptr<InsQueue>> hcclCacheMap_; //存储aiv cache信息
301 : HcclResult GetCacheMap(AivOpCacheArgs& opCacheParam , std::shared_ptr<InsQueue>& tempInsQue);
302 : HcclResult SetAccelerator(HcclAccelerator hcclAccelerator, bool isCcuMsAvailable);
303 : HcclResult GetAccelerator(int32_t* accelerator) const;
304 : void ExecAlgSelect(const CollOpParams &opParams, const OpMode &opMode);
305 :
306 : bool IsOpUsingCcuMs() const; // 算子粒度
307 : bool IsOpUsingCcuSched() const; // 算子粒度
308 : bool IsOpUsingAiv() const; // 算子粒度
309 : bool IsOpUsingAivOnly() const; // 算子粒度
310 : bool IsCommUsingCcuMs() const; // 通信域粒度
311 : bool IsCommUsingCcuSched() const; // 通信域粒度
312 : void RegisterAcceStateCallBack(std::function<HcclResult(const std::string &commId, bool isUsingCcuMs, bool isUsingCcuSched)> inCallback);
313 : HcclResult AcceleratorFallback();// 加速模式回退
314 :
315 : virtual UbMemoryTransportMgr *GetUbMemoryTransportMgr() const;
316 1 : u32 GetAivTag() const
317 : {
318 1 : return aivTag;
319 : }
320 :
321 2 : u32 GetAivOffloadTag() const
322 : {
323 2 : return aivOffloadTag;
324 : }
325 0 : u8 GetAlgorithmType() const
326 : {
327 0 : return algorithmType_;
328 : }
329 :
330 2 : void SetAivTag(u32 tag)
331 : {
332 2 : aivTag = tag;
333 2 : }
334 :
335 2 : void SetAivClearEnable(bool enable)
336 : {
337 2 : aivClearEnable = enable;
338 2 : }
339 :
340 1 : bool GetAivClearEnable() const
341 : {
342 1 : return aivClearEnable;
343 : }
344 :
345 1 : void SetAivCoreLimit(u32 newAivCoreLimit)
346 : {
347 1 : aivCoreLimit = newAivCoreLimit;
348 1 : }
349 : HcclResult CalcTaskNum(OpType opType, DataType dataType, u64 count, u32 &taskNum) const;
350 : void CollAlgComponentInit();
351 :
352 : virtual CcuStreamSyncNotifyManager &GetCcuStreamSyncNotifyManager() const;
353 :
354 15 : void saveCCUParams(std::vector<std::vector<CcuTaskParam>> &&ccuParams,
355 : std::vector<std::vector<CcuProfilingInfo>>&&ccuProfilingInfo, u64 execId, CcuInstType insType,
356 : bool isSlave = false)
357 : {
358 15 : auto &ccuParamsMapping = colCcuParamMapping[currentCollOperator->opType];
359 15 : auto &ccuParamsNotCacheKey = colParamsNotCacheKey[currentCollOperator->opType];
360 29 : if (ccuParamsMapping.find(ccuParamsMappingKey) == ccuParamsMapping.end() &&
361 29 : ccuParamsNotCacheKey.find(ccuParamsMappingKey) == ccuParamsNotCacheKey.end()) {
362 13 : ccuParamsMapping.emplace(std::piecewise_construct, std::forward_as_tuple(ccuParamsMappingKey),
363 13 : std::forward_as_tuple(std::move(ccuParams), std::move(ccuProfilingInfo), execId,
364 26 : insType, isSlave, static_cast<void *>(this)));
365 : } else {
366 2 : ccuParamsMapping.erase(ccuParamsMappingKey);
367 2 : if (ccuParamsMapping.empty()) {
368 2 : colCcuParamMapping.erase(currentCollOperator->opType);
369 : }
370 2 : ccuParamsNotCacheKey.emplace(ccuParamsMappingKey);
371 : }
372 15 : }
373 :
374 0 : inline bool isEnableSuperFasterLoad() const
375 : {
376 0 : return superFasterLoad;
377 : }
378 :
379 : HcclResult CreateBarrierMemory(void *&sendBuf, void *&recvBuf, uint64_t count);
380 :
381 : HcclResult HcomSelectAlg(const CollOpParams &opParams, int32_t aivCoreLimit, bool &ifAiv, std::string &algName);
382 : HcclResult CalcNumBlocks(const CollOpParams &opParams, int32_t aivCoreLimit, std::string &algName,
383 : u32 &numBlocks) const;
384 : HcclResult GetAlgExecParam(const CollOpParams &opParams, bool clearEnable, void *&commContext, u64 &len,
385 : u32 aivCoreLimit);
386 :
387 : HcclResult ClearOpResource(const std::string &opTag);// 清空opTag所属资源
388 : HcclResult GetAicpuOpStreamNotify(rtStream_t *opStream, u8 aicpuNotifyNum, void** aicpuNotify) const;
389 : static std::string GetTopoFilePath();
390 : std::vector<LinkData> GetFullMeshLinks() const;
391 : ErrorMessageReport GetAicpuTaskException();
392 : u32 GetRankInParentComm() const;
393 : aclrtFuncHandle GetAicpuKernelFuncHandle(const char *kernelName) const;
394 : bool IsCommWithPCIEProtocol() const; // 判断通信域内是否有rank之间存在PCIE链路
395 : HcclResult Mc2AiCpuStreamAllocAndGetV2(rtStream_t *aiCpuStream);
396 : HcclResult SaveDpuStreamId();
397 1 : uint32_t GetDpuStreamId() const {
398 1 : return dpuStreamId;
399 : }
400 :
401 : HcclResult GetRankIpPortMap(RankIpPortMapPtr& rankIpPortMap);
402 : HcclResult SetRankIpPortMap(const RankIpPortMapPtr& rankIpPortMap);
403 :
404 : private:
405 : std::string id;
406 : static std::atomic<u32> globalIndex; // 全局通信域唯一一个index, 对应锁保护
407 : u32 idIndex{0}; // 每个通信域唯一一个index
408 : std::string establishLinkSocketTag;
409 : RankId myRank;
410 : u32 rankSize;
411 : RankId rankInParentComm;
412 : DevType devType;
413 : DevId devPhyId;
414 : DevId devLogicId;
415 : HcclCommConfig config;
416 : std::shared_ptr<RankGraph> rankGraph;
417 : uint32_t dpuStreamId{0};
418 : unique_ptr<DataBufManager> dataBufferManager;
419 : unique_ptr<LocalRmaBufManager> localRmaBufManager;
420 : unique_ptr<RemoteRmaBufManager> remoteRmaBufManager;
421 : unique_ptr<QueueNotifyManager> aicpuQueueNotifyManager_;
422 : unique_ptr<QueueNotifyManager> ccuQueueNotifyManager_;
423 : unique_ptr<QueueWaitGroupCntNotifyManager> queueWaitGroupCntNotifyManager;
424 : unique_ptr<QueueBcastPostCntNotifyManager> queueBcastPostCntNotifyManager;
425 : unique_ptr<ConnLocalNotifyManager> connLocalNotifyManager;
426 : unique_ptr<ConnLocalCntNotifyManager> connLocalCntNotifyManager;
427 : unique_ptr<StreamManager> streamManager;
428 : unique_ptr<AicpuStreamManager> aicpuStreamManager;
429 : unique_ptr<SocketManager> socketManager;
430 : unique_ptr<RmaConnManager> rmaConnectionManager;
431 : CollServiceBase *collService{nullptr};
432 : unique_ptr<CollOperator> currentCollOperator;
433 : unique_ptr<HostDeviceSyncNotifyManager> hostDeviceSyncNotifyManager;
434 : unique_ptr<Trace> trace;
435 : unique_ptr<MemTransportManager> memTransportManager{};
436 : unique_ptr<MirrorTaskManager> mirrorTaskManager;
437 : unique_ptr<UbMemoryTransportMgr> ubMemoryTransportMgr{};
438 : unique_ptr<ProfilingReporter> profilingReporter;
439 : unique_ptr<HDCommunicate> kfcControlTransferH2D;
440 : unique_ptr<HDCommunicate> kfcStatusTransferD2H;
441 : unique_ptr<HcclOneSidedService> oneSidedService;
442 : std::function<void()> printChannelInfoCallback{nullptr};
443 : unique_ptr<CcuStreamSyncNotifyManager> ccuStreamSyncNotifyManager;
444 : std::shared_ptr<CcuDriverHandle> ccuDrvHandle{nullptr};
445 :
446 : std::vector<u32> netLayersVec;
447 : std::vector<uint32_t> instSizeVec;
448 : std::vector<uint32_t> rankListVec;
449 : std::vector<CommLink> linkListVec;
450 : std::vector<uint32_t> ranksVec;
451 : std::vector<uint32_t> topoInstsVec;
452 : std::vector<u32> enableP2PDevices_;
453 :
454 : NotifyTimeoutCfg notifyTimeoutCfg;
455 :
456 : u32 step = 0; // 全局device信息的step
457 : u32 opBaseOpIndex = 0; // 单算子次数
458 : u32 collOpIndex = 0; // 集合通信算子次数
459 : u32 opIndex = 0; // 下发算子总计数(单算子/图模式/CCU快速下发)
460 : u32 sendRecvIndex = 0; // send/recv 算子次数
461 : u32 submittedOpCnt = 0;
462 : u32 aivCoreLimit = MAX_NUM_BLOCKS;
463 :
464 : void RegisterOffloadSlaveStreams(const std::string &opTag, std::vector<void *> slaveStreams) const;
465 : void RegisterOffloadScratchBuffer(const std::string &opTag, void *scratchMemPtr, u64 requiredScratchMemSize);
466 : bool initFlag = false;
467 : bool devModeFlag = false;
468 : bool isSuspended = false;
469 : bool isCleaned = false;
470 : bool isAicpuKernelLaunched = false;
471 : bool isDpuKernelLaunched = false;
472 : bool isWorldGroup = false;
473 : bool aivClearEnable = false;
474 :
475 : std::shared_ptr<DevBuffer> cclBuffer;
476 : u64 cclBufferSize = 0;
477 : std::shared_ptr<DevBuffer> aivTagBuffer;
478 : std::shared_ptr<DevBuffer> indirectInCclBuffer;
479 : std::shared_ptr<DevBuffer> indirectOutCclBuffer;
480 : std::shared_ptr<DevBuffer> aivOffloadTagBuffer;
481 : std::shared_ptr<DevBuffer> inCclBuffer;
482 : std::shared_ptr<DevBuffer> outCclBuffer;
483 : // 为barrier算子新增的buffer与判断;
484 : std::shared_ptr<DevBuffer> barrierInMemory;
485 : std::shared_ptr<DevBuffer> barrierOutMemory;
486 : std::unordered_map<std::string, std::shared_ptr<DevBuffer>> tagWorkspaceMap_;
487 : bool isFirstBarrier = true;
488 : DpuKernelLaunchParam hostArgs;
489 : std::unordered_map<std::string, DpuShmem> tagDpuShmemArgsMap_ = {{"DPUTAG", DpuShmem{}},
490 : {"DPUTASKEXCEPTION", DpuShmem{}}};
491 : void* hostShareBuf{nullptr}; // 在dpu线程中，从DPUTAG共享内存拷出算法信息的存放地址，作为入参给算法注册的回调
492 : std::unordered_map<std::string, std::shared_ptr<DevBuffer>> tagWorkspaceVAMap_;
493 : aclrtStream dpuStream;
494 : aclrtContext dpuContext;
495 : aclrtContext npuContext;
496 :
497 : std::unordered_map<std::string, std::shared_ptr<Buffer>> offloadScrachBufferMap;
498 : BinaryStream staticBinaryInfo; // 静态信息序列化流
499 :
500 : std::atomic<CommStatus> status_{CommStatus::COMM_IDLE}; // 通信域状态
501 : std::vector<u32> rankIdsVec; // 子通信域使用：序列化解析
502 : std::unique_ptr<RankTableInfo> ranktableInfo; // 主通信域使用：序列化解析
503 : std::shared_ptr<TopoInfo> topoInfo; // 主通信域使用：序列化解析
504 :
505 : std::map<AcceleratorState, std::shared_ptr<CollServiceBase>> collServices; // 初始化3种collService，供算法选择
506 : std::shared_ptr<CollAlgComponent> collAlgComponent; // 初始化算法组件
507 : OpExecuteConfig opExecuteConfig; // 算子粒度 加速模式
508 : OpExecuteConfig commExecuteConfig; // 通信域粒度 加速模式
509 : std::string curAlgName; // 当前算法名称
510 : bool isLoadOp{false}; // 是否已加载过算子,只要算子下发过就不让改加速模式 loadop offload AllocCommResource
511 : u32 aivTag{1}; // aiv kernal内部用于标志位计数
512 : u32 aivOffloadTag{0};// aiv kernal内部用于标志位计数
513 : u8 algorithmType_{0};
514 : std::atomic<u32> tagResourceIndex_{0};
515 :
516 : std::function<HcclResult(const std::string &commId, bool isUsingCcuMs, bool isUsingCcuSched)> callback;
517 : CollOpParams curOpParams; // 当前算子参数
518 : std::map<std::pair<OpType, string>, std::pair<AcceleratorState, string>>
519 : opAcceStateCache{}; // opType + algName --> acceleratorState + newAlgName
520 : AicpuBinaryHolder aicpuKernelHolder_;
521 :
522 : void InitCommonData(const CommParams &commParams);
523 : void InitCommonDataNotInitDevType(const CommParams &commParams, const HcclCommConfig &commConfig);
524 : void InitCommonData(const CommParams &commParams, const HcclCommConfig &commConfig);
525 : void InitRankGraph(const string &ranktableM);
526 : void InitRankGraph(std::unique_ptr<RankGraph> &inputRankGraph);
527 : void InitRankGraph(const RankTableInfo &ranktable);
528 : void CheckRankGraph() const;
529 : void CheckRankGraphAddrs() const;
530 : HcclResult CheckCommStatus() const;
531 : void InitDataBufferManager();
532 : void InitNotifyManager();
533 : void InitStreamManager();
534 : void InitCollService();
535 : void InitHccpHdc() const;
536 : void InitCcuSuperFastLoad();
537 : void InitPreResource();
538 : void DeInitPreResource();
539 : void InitSocketManager();
540 : void InitRmaConnManager();
541 : void InitMemTransportManager();
542 : void InitHostDeviceSyncNotifyManager();
543 : HcclResult InitTraceManager();
544 : void InitHDCommunicate();
545 : void InitOneSidedService();
546 : void InitUbMemoryTransportMgr();
547 : void RefreshSubmittedOpcnt();
548 : void SingleRankProc(const CollOpParams &opParams, void *stream) const;
549 : void ConvertCollOperatorA2A(const CollOpParams &opParams, bool isLaunch = true, bool isHcomSelectAlg = false);
550 : void DefaultConvertCollOperatorA2A(const CollOpParams &opParams);
551 : void LaunchConvertCollOperatorA2A(const CollOpParams &opParams, bool isHcomSelectAlg = false);
552 : void ConvertCollOperatorMem(const CollOpParams &opParams, u64 size);
553 : void CalcA2ASendRecvMem(const CollOpParams &opParams, u64 &sendSize, u64 &recvSize, bool isHcomSelectAlg = false) const;
554 : void ConvertCollOperatorMemV(const CollOpParams &opParams, bool isHcomSelectAlg = false);
555 : void RegisterAicpuKernel();
556 :
557 : // dpu相关
558 : void InitHccpPeer() const; // 拉起peer模式HCCP进程
559 : bool IsNeedDpu(); // 判断是否需要Host网卡参与集合通信
560 : void InitDpuKernel();
561 : std::unordered_set<IpAddress> GetHostIpFromRankGraph();
562 : HcclResult LaunchDpuKernel(aclrtFuncHandle &funcHandle);
563 : HcclResult PrepareDpuKernelResource(aclrtFuncHandle &funcHandle);
564 : HcclResult DestroyDpuKernelResource();
565 : HcclResult WaitDpuKernelThreadTerminate();
566 : HcclResult InitAndLaunchDpuKernel();
567 : HcclResult InitAndLaunchAicpuKernel(); // 保存dpu taskexception共享内存到aicpu全局map中
568 : HcclResult DestroyDpuTaskexpShmemInDevice();
569 :
570 : HcclResult Init(const CommParams &commParams, std::unique_ptr<RankGraph> &inputRankGraph, DevId inputDevLogicId);
571 : HcclResult Init(const CommParams &commParams, std::unique_ptr<RankGraph> &inputRankGraph,
572 : HcclCommConfig &subConfig, DevId inputDevLogicId);
573 : HcclResult InitCommResource(const CommParams &commParams);
574 :
575 : void WaitReady() const;
576 :
577 : void InitMirrorTaskManager();
578 : HcclResult InitProfilingReporter();
579 : void UpdateProfStat();
580 : void InitTaskExceptionHandler() const;
581 :
582 : // 配置cachedReq属性：静态图模式：true, 动态图模式：false， 单算子模式：false
583 : void ReportProfInfo(uint64_t beginTime, bool cachedReq, bool opbased);
584 :
585 : void SelectCollService(); // 根据配置选择对应的collService
586 :
587 : CcuSFLMappingKey ccuParamsMappingKey{};
588 : std::unordered_map<const OpType, std::unordered_map<CcuSFLMappingKey, CachedCCUParams, ArrayHasher>>
589 : colCcuParamMapping{};
590 : std::unordered_map<const OpType, std::unordered_set<CcuSFLMappingKey, ArrayHasher>> colParamsNotCacheKey{};
591 : bool superFasterLoad{false};
592 : bool taskExceptionEnv{true}; // 默认HCCL_DFS_CONFIG="task_exception:on" 且默认on下不开启快速下发
593 : bool enableProfilingEnv{false};
594 : bool TryFastCcuLaunch(const CollOpParams &opParams, aclrtStream const stream);
595 : void FillAllToAllVArgs(const CollOpParams &opParams, rtCcuTaskInfo_t *&ccuParams) const;
596 : void ExecuteFastCcuLaunch(const CollOpParams &opParams, aclrtStream const stream, CachedCCUParams ¶ms);
597 :
598 : void OpAcceleratorStateFallback(); // 算子粒度加速模式状态回退
599 : HcclResult ReLoadOpbasedOp();
600 : HcclResult ReLoadOffloadOp();
601 :
602 : void TryInitCcuFeature() const; // 根据通信域加速模式和rank信息，选择打开ccu功能，依赖hdc通道
603 :
604 : template<typename BufferType>
605 : static std::shared_ptr<BufferType> BarrierAllocBuffer(std::size_t size);
606 :
607 : void AppendLocalDieIdForLinks();
608 : HcclResult SetAivControledCoreNum(bool isAiv);
609 :
610 : void CheckAcceleratorConsistency(AcceleratorState commAccelerator, AcceleratorState tilingAccelerator) const;
611 : HcclResult GetTilingAccelerator(void *mc2Tiling, AcceleratorState& acceleratorState) const;
612 :
613 : // AICPU场景aclgraph专用
614 : HcclResult OffloadResourcePre(std::string &opTag, const CollOpParams &opParams);
615 :
616 : RankIpPortMapPtr rankIpPortMap_;
617 : };
618 : } // namespace Hccl
619 :
620 : #endif // HCCL_COMMUNICATOR_IMPL_H
|
