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