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_IMPL_H
12 : #define HCCL_IMPL_H
13 :
14 : #include <functional>
15 : #include <vector>
16 : #include <hccl/hccl_types.h>
17 :
18 : #include "hccl_common.h"
19 : #include "common.h"
20 : #include "comm_factory_pub.h"
21 : #include "parallel_task_loader.h"
22 : #include "dispatcher.h"
23 : #include "ccl_buffer_manager.h"
24 : #include "workspace_resource.h"
25 : #include "hccl_impl_pub.h"
26 : #include "op_base_stream_manager_pub.h"
27 : #include "resource_manager/queue_notify_manager.h"
28 : #include "device_capacity.h"
29 : #include "coll_alg_utils.h"
30 : #include "alg_configurator.h"
31 : #include "workflow_pub.h"
32 : #include "topo_info_extractor.h"
33 :
34 : namespace hccl {
35 : constexpr s32 COMM_INDEX_0 = 0;
36 : constexpr s32 COMM_INDEX_1 = 1;
37 : constexpr s32 STREAM_INDEX_0 = 0;
38 : constexpr s32 STREAM_INDEX_1 = 1;
39 : constexpr s32 COMM_SIZE_TWO = 2;
40 : constexpr s32 LEVEL1_PLANE_NUM_IN_4PMESH = 4;
41 : constexpr s32 LEVEL0_PLANE_NUM_IN_NPRING_SINGLE = 1;
42 : constexpr s32 LEVEL0_PLANE_NUM_IN_NPRING_DOUBLE = 2;
43 : constexpr s32 RDMA_PLANE_NUM_IN_NPRING_DOUBLE = 2;
44 : constexpr s32 LEVEL0_PLANE_NUM_IN_8PRING = 4;
45 : constexpr s32 LEVEL0_PLANE_NUM_IN_4PMESH = 3;
46 : constexpr s32 STREAM_NUM_FOR_DMAREDUCE_ONE_RING = 2;
47 :
48 : constexpr u32 SLICES_FACTOR = 2;
49 : constexpr u32 RDMA_ADD_STREAMS_NUM = 3;
50 :
51 : constexpr u32 CCE_REDUCE_ALIGN_SIZE = 32;
52 :
53 : constexpr u32 HCCL_INTERNODE_MAX_DATA_RATE = 1; // node间通信的单次通信量最多为node通信量的1倍（R-HD或NHR）
54 :
55 : constexpr u32 DEVICE_EIGHT = 8;
56 : constexpr u32 DEVICE_FOUR = 4;
57 : constexpr u32 DEVICE_TWO = 2;
58 : constexpr u32 DEVICE_ONE = 1;
59 : constexpr u64 KB2B = 1024;
60 : constexpr u64 FACTOR_TWO = 2;
61 : constexpr float BASE_COMM_LATENCY = 13.0; // 通信静态时延（us）;
62 : constexpr u64 MULTIPLIER_MS2US = 1000; // 毫秒转换成微秒
63 : constexpr u64 MULTIPLIER_S2MS = 1000; // 秒转换成毫秒
64 :
65 : using ResDeviceMemMap = std::map<std::string, DeviceMem>;
66 :
67 : struct PiplineSliceInfo {
68 : std::vector<Slice> piplineDataSegsSlice;
69 : std::vector<std::vector<Slice>> piplineMultiStreamSlice;
70 : u64 count{0};
71 : u64 offset{0}; // 记录切分内存段起始地址的相对偏移
72 : };
73 :
74 : class hcclImpl {
75 : friend class CollAlgOperator;
76 : friend class CollNativeExecutorBase;
77 :
78 : public:
79 : explicit hcclImpl(
80 : const HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool>& notifyPool,
81 : std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap,
82 : const std::unique_ptr<QueueNotifyManager>& queueNotifyManager, std::unique_ptr<WorkspaceResource>& workSpaceRes,
83 : CCLBufferManager& cclBufferManager, HcclAlgoAttr& algoAttr, HcclTopoAttr& topoAttr,
84 : std::shared_ptr<AlgConfigurator> algConfigurator, std::shared_ptr<TopoInfoExtractor> topoInfoEx);
85 : ~hcclImpl();
86 : HcclResult Init(bool isHeterogComm = false);
87 : HcclResult ReleaseCommInfos();
88 : HcclResult CreateMutiStreamRes(
89 : const std::string& tag, Stream& stream, AlgType algType, bool isBatchSendRecv = false, u32 ringNum = 0);
90 :
91 : HcclResult CreateMutiStreamRes(
92 : const std::string& tag, Stream& stream, level1StreamInfo_t& streamInfo, AlgType algType,
93 : bool isAicpuModeEn = false, bool isBatchSendRecv = false, u32 ringNum = 0);
94 : HcclResult CreateComm(
95 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, AlgType algType,
96 : std::unique_ptr<CommInfo>& commInfo, u32 root = INVALID_VALUE_RANKID, bool isP2p = false,
97 : bool isAicpuModeEn = false, bool isBatchSendRecv = false, bool meshSinglePlane = false, bool aivMode = false,
98 : std::set<u32> batchSendRecvtargetRanks = std::set<u32>());
99 :
100 : HcclResult CreateComm(
101 : const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem, AlgType algType,
102 : u32 root = INVALID_VALUE_RANKID, bool isP2p = false, bool isBatchSendRecv = false, bool meshSinglePlane = false,
103 : bool aivMode = false, std::set<u32> batchSendRecvtargetRanks = std::set<u32>());
104 : HcclResult ClearOpResource(const std::string& tag);
105 0 : void Break()
106 : {
107 0 : if (Is310P3Common(isHaveCpuRank_, deviceType_)) {
108 0 : HCCL_ERROR("[hcclImpl][Break]Break is not supported.");
109 0 : return;
110 : }
111 0 : for (auto& commInfo : tagCommInfo_) {
112 0 : for (auto& comm : commInfo.second.commLevel0) {
113 0 : if (comm == nullptr) {
114 0 : continue;
115 : }
116 0 : comm->Break();
117 : }
118 0 : for (auto& comm : commInfo.second.commLevel1) {
119 0 : if (comm == nullptr) {
120 0 : continue;
121 : }
122 0 : comm->Break();
123 : }
124 0 : for (auto& comm : commInfo.second.commP2P) {
125 0 : if (comm == nullptr) {
126 0 : continue;
127 : }
128 0 : comm->Break();
129 : }
130 : }
131 : }
132 :
133 0 : inline bool IsExistCommRes(const std::string& tag)
134 : {
135 0 : std::unique_lock<std::mutex> commLock(commLock_);
136 0 : return (tagCommInfo_.find(tag) != tagCommInfo_.end());
137 0 : }
138 :
139 : void SetHDCModeInfo(
140 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>>& rankDevicePhyIdNicInfoMap,
141 : std::vector<u32>& ranksPort, std::vector<u32>& vnicRanksPort, bool isSetHDCModeInfo, bool isUseRankPort);
142 :
143 : private:
144 : void SetAlgoAttr(HcclAlgoAttr& algoAttr);
145 : void SetTopoAttr(HcclTopoAttr& algoAttr);
146 : HcclResult CreateCommThread(
147 : const ErrContextPub& error_context, const std::string& tag, DeviceMem& inputMem, DeviceMem& outputMem,
148 : DeviceMem& expMem, const CommParaInfo& commParaInfo, std::vector<std::unique_ptr<CommBase>>& commVec,
149 : HcclResult& retOut);
150 : HcclResult GetCommTypeInLevel0(const AlgType algType, const TopoType topoType, CommType& commType);
151 : HcclResult GetCommTypeInLevel1(const AlgType algType, CommType& commType);
152 : CommPlane GetCommPlaneInLevel1(CommType& commType);
153 : HcclResult ReplaceCommInfoByTag(const std::string& tag, std::unique_ptr<CommInfo>& commInfo);
154 : HcclResult CreateP2pComm(const std::string& tag, CommInfo& commInfo, DeviceMem& inOutMem, u32 peerUserRank);
155 : HcclResult CreateCommByAlg(
156 : const std::string& tag, const AlgType algType, CommInfo& commInfo, DeviceMem& inputMem, DeviceMem& outputMem,
157 : DeviceMem& expMem, u32 root = INVALID_VALUE_RANKID, bool isAicpuModeEn = false, bool meshSinglePlane = false,
158 : bool isA2MC2MultiServer = false);
159 :
160 : void DestroyLevel1Comm(const std::string& tag);
161 : void DestroyIntraServerComm(const std::string& tag);
162 : void DestroyLevel0Comm(const std::string& tag);
163 : HcclResult ReleaseSignal(level1StreamInfo_t& level1Stream);
164 :
165 : HcclResult InitMultiStreamResource(
166 : const std::string& tag, level1StreamInfo_t& streamInfo, AlgType algType, bool isAicpuModeEn = false,
167 : bool isBatchSendRecv = false, u32 ringNum = 0);
168 :
169 : HcclResult WaitCommThread(std::unique_ptr<std::thread>& ThreadPtr) const;
170 : HcclResult SetRankPortInfo(s32 deviceLogicID, bool isUseRankPort, std::vector<u32>& ranksPort);
171 :
172 : /* ---------------以下为私有成员变量定义领域-------------------------- */
173 : TopoType topoType_ = TopoType::TOPO_TYPE_COMMON;
174 : std::mutex commLock_;
175 :
176 : tagCommInfo_t tagCommInfo_; // 以tag为粒度分配comm实例和资源
177 : std::mutex tagStreamInfoLock_;
178 : std::mutex scratchMemLock_;
179 : std::map<std::string, DeviceMem> scratchMemMap_;
180 : std::vector<u32> nicList_;
181 : std::mutex nicSendSizeListLock_;
182 : std::map<std::string, std::vector<u64>> nicSendSizeList_;
183 : std::mutex ringNicListLock_;
184 : std::map<std::string, std::vector<std::vector<u32>>> ringNicList_;
185 : u32 serverNum_;
186 : u32 superPodNum_;
187 : u32 moduleNum_;
188 : OpBaseStreamManager opBaseStreamManager_;
189 : std::vector<Stream> auxRingStreamsDev_;
190 :
191 : std::unique_ptr<std::thread> commThreadPtrLevel0_;
192 : std::unique_ptr<std::thread> commThreadPtrLevel0Rdma_;
193 : std::unique_ptr<std::thread> commThreadPtrLevel1_;
194 : std::unique_ptr<std::thread> commThreadPtrLevel1Rdma_;
195 : std::unique_ptr<std::thread> commThreadPtrLevel2_;
196 :
197 : std::unique_ptr<CommBase> commMeshPtr_; // 单算子alltoallv只建链一次
198 : std::unique_ptr<CommBase> commMeshLevel2_; // 单算子alltoallv只建链一次
199 : std::map<std::string, std::unique_ptr<CommBase>> commMeshMap_; // 图模式alltoallv建链多次
200 : u32 deviceNumPerServer_;
201 : u32 deviceNumPerAggregation_;
202 : static std::array<DeviceMem, MAX_MODULE_DEVICE_NUM> inOutPutTempMem_; // 图模式alltoallv输入为0时用该内存建链
203 : static std::array<std::mutex, MAX_MODULE_DEVICE_NUM> inOutPutTempMemMutex_;
204 : static std::array<Referenced, MAX_MODULE_DEVICE_NUM> instanceRef_; // 实例计数，用于释放静态资源
205 : const u64 tinyMemSizeForTransportCreation{LARGE_PAGE_MEMORY_MIN_SIZE}; // 避免申请小页内存。最小2*1024*1024
206 : bool isAlltoAllZCopyMode_ = false;
207 : bool needRecreateAlltoallComm_ = false;
208 : std::map<std::string, bool> isAlltoAllZCopyModeMap_;
209 : // 按照 tag 记录全局所有卡上 alltoall 算子的中转内存大小
210 : std::unordered_map<std::string, std::unordered_map<u32, u64>> allRankAlltoallScratchMemSize_;
211 : bool isSingleMeshAggregation_ = false;
212 : bool meshSinglePlane_ = false;
213 : bool isAllRankSamePlane_ = false;
214 :
215 : u64 piplineSliceNum_ = 0; // Server间pipline切分数量 0: 不支持; 1: 当前数据量下切1份; 其他: 走pipline模式
216 : const HcclDispatcher dispatcher_; // dispatcher放到最后析构
217 : const std::unique_ptr<NotifyPool>& notifyPool_;
218 : std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap_;
219 : const std::unique_ptr<QueueNotifyManager>& queueNotifyManager_;
220 : std::unordered_map<u32, u32> pairLinkCounter_; // server内所有device间的链路类型计数
221 : std::unordered_map<u32, std::unordered_map<int, std::vector<int>>> pairLinkInfo_; // server内所有device间的链路类型
222 : bool isHaveCpuRank_;
223 : u32 userRank_; // 本group中的userrank
224 : u32 realUserRank_; // world group中的userrank
225 : u32 userRankSize_;
226 : std::vector<RankInfo> rankInfoList_; // world group内rank的信息, 按照rank id递增依次排列
227 : bool inlineReduceSwitchOn_;
228 : NICDeployment nicDeployment_;
229 : u32 devicePhyId_;
230 : s32 deviceLogicId_;
231 : bool isUsedRdmaLevel0_;
232 : std::unique_ptr<WorkspaceResource>& workSpaceRes_;
233 : CCLBufferManager& cclBufferManager_;
234 : DevType deviceType_;
235 : std::string collectiveId_;
236 : std::unique_ptr<CommFactory> commFactory_;
237 : WorkMode commWorkMode_;
238 : tagStreamInfo_t tagStreamInfo_;
239 : u32 meshAggregationRankSize_;
240 : std::string identifier_;
241 : bool isDiffDeviceModule_;
242 : bool isStandardCard_;
243 : bool is310PDuoCard_;
244 : bool multiModuleDiffDeviceNumMode_;
245 : bool multiSuperPodDiffServerNumMode_;
246 : bool multiSuperPodDiffDeviceNumMode_;
247 : bool isUsedInterHccsMode_ = false;
248 : bool useSuperPodMode_ = false;
249 : s32 pid_ = 0;
250 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>> rankDevicePhyIdNicInfoMap_{};
251 : std::vector<u32> ranksPort_;
252 : bool isSetHDCModeInfo_{false};
253 : bool isUseRankPort_{false};
254 : std::vector<u32> vnicRanksPort_;
255 : bool isSupportRdmaLite_{false}; // 是否支持rdma lite
256 : bool isSupportHccsAndSio_{false}; // 是否支持hccs sio并发
257 :
258 : HcclWorkflowMode workflowMode_{HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE};
259 : std::shared_ptr<AlgConfigurator> algConfigurator_;
260 : std::shared_ptr<TopoInfoExtractor> topoInfoEx_;
261 : HcclTopoAttr& topoAttr_;
262 : HcclAlgoAttr& algoAttr_;
263 : u32 localNicPort_{0};
264 : bool isNeedInitNic_{false};
265 : };
266 : } // namespace hccl
267 :
268 : #endif /** __HCCL_COMM_H__ */
|
