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