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 TRANSPORT_MANAGER_H
12 : #define TRANSPORT_MANAGER_H
13 :
14 : #include <mutex>
15 : #include <unordered_map>
16 : #include <atomic>
17 : #include <fstream>
18 : #include "hccl/base.h"
19 : #include "hccl_socket_manager.h"
20 : #include "dispatcher.h"
21 : #include "mem_device_pub.h"
22 : #include "transport_pub.h"
23 : #include "ccl_buffer_manager.h"
24 : #include "externalinput_pub.h"
25 : #include "sal_pub.h"
26 : #include "thread/threads_guard.h"
27 : #include "hccl_hash_utils.h"
28 : #include "workflow_pub.h"
29 : #include "comm_base_pub.h"
30 : #include "coll_alg_param.h"
31 : #include "multi_qpInfo_manager.h"
32 : namespace hccl {
33 :
34 : constexpr u32 AICPU_RETRY_BACKUP_PORT = 16667; // aicpu重执行备份默认端口
35 : constexpr u32 MASSIVE_IBV_CONNECTION_COUNT = 1000; // bsr大于这个链路数量就切换链路类型
36 : constexpr u32 SEND_QP_DEPTH_FOR_BSR = 512; // 使用Transport NpuDriect链路的时候设置send深度为512
37 : constexpr u32 RECV_QP_DEPTH_FOR_BSR = 128; // // 使用Transport NpuDriect链路的时候设置recv深度为128
38 : constexpr u32 MAX_THREAD_NUM = 8; // BatchSendRecv建链时单个线程池的最大线程数量
39 :
40 : struct TransportData {
41 : LinkMode linkMode{LinkMode::LINK_RESERVED_MODE};
42 : std::vector<HcclIpAddress> remoteIpAddr;
43 : u32 remoteUserrank{INVALID_VALUE_RANKID};
44 : u32 remoteWorldRank{INVALID_VALUE_RANKID};
45 : s32 remoteDeviceId{-1};
46 : DevType deviceType{DevType::DEV_TYPE_COUNT};
47 : DeviceMem inputMem{DeviceMem()};
48 : DeviceMem outputMem{DeviceMem()};
49 : bool supportDataReceivedAck{false};
50 : u32 remoteSocketPort;
51 :
52 : TransportData(
53 : LinkMode linkMode, const std::vector<HcclIpAddress>& remoteIpAddr, u32 remoteUserrank, u32 remoteWorldRank,
54 : s32 remoteDeviceId, DevType deviceType, const DeviceMem& inputMem, const DeviceMem& outputMem,
55 : bool supportDataReceivedAck, u32 remoteSocketPort)
56 : : linkMode(linkMode),
57 : remoteIpAddr(remoteIpAddr),
58 : remoteUserrank(remoteUserrank),
59 : remoteWorldRank(remoteWorldRank),
60 : remoteDeviceId(remoteDeviceId),
61 : deviceType(deviceType),
62 : inputMem(inputMem),
63 : outputMem(outputMem),
64 : supportDataReceivedAck(supportDataReceivedAck),
65 : remoteSocketPort(remoteSocketPort) {};
66 :
67 : bool operator==(const TransportData& that) const
68 : {
69 : return (linkMode == that.linkMode) && (remoteIpAddr == that.remoteIpAddr)
70 : && (remoteUserrank == that.remoteUserrank) && (remoteWorldRank == that.remoteWorldRank)
71 : && (remoteDeviceId == that.remoteDeviceId) && (deviceType == that.deviceType)
72 : && (inputMem == that.inputMem) && (outputMem == that.outputMem)
73 : && (supportDataReceivedAck == that.supportDataReceivedAck)
74 : && (remoteSocketPort == that.remoteSocketPort);
75 : }
76 : };
77 :
78 : struct SubCommLinkPara {
79 : struct SingleSubCommTransport& singleSubCommTransport;
80 : std::vector<std::pair<u32, u32>> remoteRankMap;
81 : u32 remoteRankIdStartIndex;
82 : u32 remoteRankIdNum;
83 : std::vector<std::unique_ptr<std::thread>> linkThreads;
84 : std::vector<HcclResult> linkResult; // TransportManager::CreateLink返回值出参
85 :
86 0 : SubCommLinkPara(
87 : struct SingleSubCommTransport& singleSubCommTransport, std::vector<std::pair<u32, u32>>& remoteRankMap,
88 : u32 remoteRankIdStartIndex, u32 remoteRankIdNum)
89 0 : : singleSubCommTransport(singleSubCommTransport),
90 0 : remoteRankMap(remoteRankMap),
91 0 : remoteRankIdStartIndex(remoteRankIdStartIndex),
92 0 : remoteRankIdNum(remoteRankIdNum)
93 0 : {}
94 :
95 0 : ~SubCommLinkPara()
96 : {
97 0 : for (auto& linkThread : linkThreads) {
98 0 : if (linkThread != nullptr && linkThread->joinable()) {
99 0 : linkThread->join();
100 : }
101 : }
102 0 : }
103 : };
104 :
105 : struct LinkPoolPara {
106 : struct SingleSubCommTransport& singleSubCommTransport;
107 : std::string poolName;
108 : // 记录pair<remoteRank, idx>, idx表示remoteRank对应的建链信息在transportRequests中的索引位置
109 : std::vector<std::pair<u32, u32>> taskList;
110 :
111 : std::atomic<u32> taskIndex{0};
112 : std::atomic<bool> abortFlag{false};
113 :
114 : std::vector<std::unique_ptr<std::thread>> linkThreads;
115 : std::vector<HcclResult> linkResults;
116 :
117 6 : LinkPoolPara(
118 : struct SingleSubCommTransport& transport, const std::string& name,
119 : const std::vector<std::pair<u32, u32>>& tasks)
120 6 : : singleSubCommTransport(transport),
121 6 : poolName(name),
122 6 : taskList(tasks)
123 : {
124 6 : u32 threadNum = std::min(MAX_THREAD_NUM, static_cast<u32>(taskList.size()));
125 6 : linkThreads.resize(threadNum);
126 6 : linkResults.resize(taskList.size(), HCCL_SUCCESS);
127 6 : }
128 :
129 6 : ~LinkPoolPara()
130 : {
131 8 : for (auto& linkThread : linkThreads) {
132 2 : if (linkThread != nullptr && linkThread->joinable()) {
133 2 : linkThread->join();
134 : }
135 : }
136 6 : }
137 : };
138 : } // namespace hccl
139 :
140 : namespace std {
141 :
142 : template <>
143 : class hash<hccl::TransportData> {
144 : public:
145 : size_t operator()(const hccl::TransportData& transportData) const
146 : {
147 : auto linkMode = hash<s32>{}(static_cast<s32>(transportData.linkMode));
148 : auto remoteIpAddrFamily = hash<s32>{}(transportData.remoteIpAddr[0].GetFamily());
149 : auto remoteIpAddr = hash<string>{}(string(transportData.remoteIpAddr[0].GetReadableAddress()));
150 : auto remoteUserrank = hash<u32>{}(transportData.remoteUserrank);
151 : auto remoteWorldRank = hash<u32>{}(transportData.remoteWorldRank);
152 : auto remoteDeviceId = hash<s32>{}(transportData.remoteDeviceId);
153 : auto deviceType = hash<s32>{}(static_cast<s32>(transportData.deviceType));
154 : auto inputMemPtr = hash<u64>{}(reinterpret_cast<u64>(transportData.inputMem.ptr()));
155 : auto inputMemSize = hash<u64>{}(transportData.inputMem.size());
156 : auto outputMemPtr = hash<u64>{}(reinterpret_cast<u64>(transportData.outputMem.ptr()));
157 : auto outputMemSize = hash<u64>{}(transportData.outputMem.size());
158 : auto supportDataReceivedAck = hash<bool>{}(transportData.supportDataReceivedAck);
159 : auto remoteSocketPort = hash<u32>{}(transportData.remoteSocketPort);
160 :
161 : return hccl::HashCombine(
162 : {linkMode, remoteIpAddrFamily, remoteIpAddr, remoteUserrank, remoteWorldRank, remoteDeviceId, deviceType,
163 : inputMemPtr, inputMemSize, outputMemPtr, outputMemSize, supportDataReceivedAck, remoteSocketPort});
164 : }
165 : };
166 : } // namespace std
167 :
168 : namespace hccl {
169 : // 独立算子内存
170 : struct IndOpMem {
171 : std::vector<HostMem> userHostMem;
172 : std::vector<DeviceMem> userDeviceMem;
173 : };
174 :
175 : struct TransportIOMem {
176 : DeviceMem cclInputMem;
177 : DeviceMem cclOutputMem;
178 : DeviceMem paramInputMem;
179 : DeviceMem paramOutputMem;
180 : DeviceMem scratchMem;
181 : DeviceMem aivInputMem;
182 : DeviceMem aivOutputMem;
183 : DeviceMem expMem;
184 : DeviceMem userMem;
185 : IndOpMem indOpMem;
186 : };
187 :
188 : class TransportManager {
189 : public:
190 : TransportManager(
191 : CCLBufferManager& cclBufferManager, const std::unique_ptr<HcclSocketManager>& socketManager,
192 : HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool>& notifyPool,
193 : const std::vector<RankInfo>& rankInfoList, RankId userRank, const std::string& identifier, s32 deviceLogicId,
194 : NICDeployment nicDeployment, bool isHaveCpuRank, bool isUseRankPort, bool isUsedRdmaLevel0,
195 : const std::vector<u32>& nicRanksPort, const std::vector<u32>& vnicRanksPort, bool useSuperPodMode,
196 : const std::vector<HcclIpAddress>& devIpAddr, const HcclIpAddress& hostIp, const HcclIpAddress& localVnicIp,
197 : std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap);
198 :
199 : ~TransportManager();
200 :
201 : HcclResult CreateVirturalTransport(SingleSubCommTransport& singleSubCommTransport);
202 : HcclResult Alloc(
203 : const std::string& tag, const TransportIOMem& transMem, OpCommTransport& opTransportResponse,
204 : bool isAicpuModeEn, bool isBackup = false, bool isZeroCopy = false,
205 0 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isCapture = false, bool isIndOp = false,
206 : bool isNpuDirectRoce = false, const OpParam* opParam = nullptr);
207 : HcclResult IncreAlloc(
208 : const std::string& tag, const TransportIOMem& transMem, OpCommTransport& opTransportReq,
209 : OpCommTransport& opTransportResponse, bool isAicpuModeEn, bool isBackup = false, bool isCapture = false,
210 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID);
211 : HcclResult
212 : GetRemoteRankList(OpCommTransport& opTransportResponse, std::vector<u32>& rankList, TransportType transportType);
213 : HcclResult
214 : GetIncreRemoteRankList(OpCommTransport& opTransportReq, std::vector<u32>& rankList, TransportType transportType);
215 : HcclResult AddremoteUserRankToList(
216 : TransportRequest& transportRequest, std::vector<u32>& rankList, TransportType transportType);
217 : TransportManager(TransportManager const&) = delete; // Copy construct
218 : TransportManager(TransportManager&&) = delete; // Move construct
219 : TransportManager& operator=(TransportManager const&) = delete; // Copy assign
220 : TransportManager& operator=(TransportManager&&) = delete; // Move assign
221 : void SetQpQosAttr(u32 trafficClass, u32 serviceLevel); // 设置TC/SL配置
222 :
223 : HcclResult SetStopFlag(bool value);
224 : bool GetStopFlag();
225 : void SetIsStandardCard(bool isStandardCard);
226 :
227 : void SetPortConfig(bool devPortSwitchOn);
228 : HcclResult CheckLinkNumAndSwitchLinkType(
229 : TransportType& type, MachinePara& machinePara, const std::vector<std::shared_ptr<HcclSocket>> sockets);
230 : void SetOpType(HcclCMDType opType);
231 : HcclResult SetGroupMode(bool groupMode);
232 : std::map<u32, TransportType> GetRemoteTransportMap();
233 :
234 : private:
235 : HcclResult GetIOMem(
236 : const TransportIOMem& transMem, const TransportMemType inputMemType, const TransportMemType outputMemType,
237 : DeviceMem& inputMem, DeviceMem& outputMem, DeviceMem& expMem);
238 : u32 GetHostPort(s32 devicePhyId);
239 : u32 GetRemoteNicPort(s32 devicePhyId, u32 dstUserRank, bool isInterRdma);
240 : bool IsSupportInterHccs(const u32 dstRank);
241 : void UpdateIsInterRdma(const u32 remoteRank, bool& isInterRdma, bool forceRdma);
242 : HcclResult
243 : MakeRemoteLinkInfo(const u32 remoteRank, bool isInterRdma, u32 socketsPerLink, HcclRankLinkInfo& remoteLinkInfo);
244 : HcclResult CreateDestSockets(
245 : const std::string& tag, RankId remoteRank, u64 taskNum,
246 : std::vector<std::shared_ptr<HcclSocket>>& connectSockets, HcclNetDevCtx& netDevCtx, bool& isInterRdma,
247 : bool forceRdma = false, bool isBackup = false, u32 subCommIndex = 0,
248 : TransportLinkType linkType = TransportLinkType::RESERVED);
249 : u32 GetSocketsPerLink(u64 taskNum, u32 remoteRankId = INVALID_VALUE_RANKID);
250 : HcclResult SetMachinePara(
251 : const std::string& tag, MachineType machineType, const std::string& serverId, u32 dstRank,
252 : const bool supportDataReceivedAck, const LinkMode linkMode,
253 : const std::vector<std::shared_ptr<HcclSocket>>& socketList, const DeviceMem& inputMem,
254 : const DeviceMem& outputMem, const DeviceMem& expMem, bool isAicpuModeEn, bool isBackup, bool isCapture,
255 : u32 notifyNum, u32 trafficClass, u32 serviceLevel, MachinePara& machinePara, RankInfo& loaclRank,
256 : RankInfo& remoteRank, const HcclNetDevCtx& netDevCtx, TransportLinkType linkType = TransportLinkType::RESERVED,
257 : const IndOpMem& indOpMem = IndOpMem(), bool isIndOp = false,
258 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isNpuDirectRoce = false);
259 : HcclResult GetTransportType(const u32 dstRank, bool isUsedRdma, TransportType& transportType);
260 : void SetTransportParam(TransportPara& para);
261 : HcclResult
262 : TransportInit(MachinePara& machinePara, std::shared_ptr<Transport>& link, bool useOneDoorbell, TransportType type);
263 : HcclResult CreateLink(
264 : const std::string& tag, const ErrContextPub& error_context, const MachineType machineType,
265 : const std::string& serverId, const u32 remoteRank, const bool supportDataReceivedAck, const LinkMode linkMode,
266 : const bool enableUseOneDoorbell, const std::string threadStr,
267 : const std::vector<std::shared_ptr<HcclSocket>> sockets, const DeviceMem inputMem, const DeviceMem outputMem,
268 : bool isUsedRdma, std::shared_ptr<Transport>& link, bool isAicpuModeEn, HcclResult& retOut,
269 : const HcclNetDevCtx& netDevCtx, u32 notifyNum = 0, bool isBackup = false, bool isCapture = false,
270 : const DeviceMem expMem = DeviceMem(), TransportLinkType linkType = TransportLinkType::RESERVED,
271 : bool isIndOp = false, const IndOpMem indOpMem = IndOpMem(),
272 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isNpuDirectRoce = false);
273 : bool IsHccsTransport(u32 remoteRank, TransportLinkType linkType);
274 : HcclResult ConstructTransTag(
275 : const std::string& tag, std::string& transTag, bool isInterRdma, u32 subCommIndex = 0, bool isHccs = false);
276 : HcclResult ExceptionHandle(const std::string& tag, OpCommTransport& opTransportResponse);
277 : HcclResult createSubCommLinkThreads(
278 : const std::string& tag, const TransportIOMem& transMem, struct SubCommLinkPara& subCommLinkPara,
279 : bool isAicpuModeEn, bool isBackup, u32 subCommIndex, bool isCapture = false,
280 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isIndOp = false, bool isNpuDirectRoce = false);
281 : HcclResult waitSubCommLinkThreadsComplete(struct SubCommLinkPara& subCommLinkPara);
282 : HcclResult
283 : checkSubCommLinkThreadsStatus(const std::string& tag, struct SubCommLinkPara& subCommLinkPara, bool isBackup);
284 : HcclResult AllocSubCommLinks(
285 : const std::string& tag, const TransportIOMem& transMem, struct SingleSubCommTransport& singleSubCommTransport,
286 : bool isAicpuModeEn, bool isBackup, u32 subCommIndex, bool isCapture = false,
287 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isIndOp = false, bool isNpuDirectRoce = false);
288 : HcclResult IsInterServer(const u32 dstRank, bool& isInterServer);
289 : HcclResult PrintErrorInfo(NicType nicType);
290 : uint32_t GetConnectMode(RankId remoteRank);
291 : HcclResult GetTransNewTag(
292 : const std::string& tag, std::string& newTag, RankId remoteRank, bool& isInterRdma, u32 subCommIndex,
293 : TransportLinkType linkType, HcclRankLinkInfo remoteLink, uint32_t mode);
294 : HcclResult CreateBatchSendRecvLinks(
295 : const std::string& tag, const TransportIOMem& transMem, struct LinkPoolPara& linkPoolPara, bool isAicpuModeEn,
296 : bool isBackup, u32 subCommIndex, bool isCapture = false,
297 : const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isIndOp = false);
298 : HcclResult WaitBatchSendRecvThreadsComplete(struct LinkPoolPara& linkPoolPara);
299 : HcclResult CheckBatchSendRecvLinkStatus(
300 : const std::string& tag, struct SingleSubCommTransport& singleSubCommTransport, bool isBackup);
301 : HcclResult AllocBatchSendRecvLinks(
302 : HcclSendRecvItem* sendRecvItemsPtr, u32 itemNum, const std::string& tag, const TransportIOMem& transMem,
303 : struct SingleSubCommTransport& singleSubCommTransport, bool isAicpuModeEn, bool isBackup, u32 subCommIndex,
304 : bool isCapture = false, const HcclCMDType& opType = HcclCMDType::HCCL_CMD_INVALID, bool isIndOp = false);
305 : HcclResult PrepareTaskLists(
306 : HcclSendRecvItem* sendRecvItemsPtr, u32 itemNum, const SingleSubCommTransport& singleSubCommTransport,
307 : std::vector<std::pair<u32, u32>>& senderList, std::vector<std::pair<u32, u32>>& receiverList);
308 :
309 : std::mutex mutex_; // 用于控制互斥资源的访问
310 : CCLBufferManager& cclBufferManager_;
311 : const std::unique_ptr<HcclSocketManager>& socketManager_;
312 : HcclDispatcher dispatcher_;
313 : const std::unique_ptr<NotifyPool>& notifyPool_;
314 : const std::vector<RankInfo>& rankInfoList_;
315 : RankId userRank_;
316 : std::string identifier_;
317 : s32 deviceLogicId_;
318 : NICDeployment nicDeployment_;
319 : bool isHaveCpuRank_{false};
320 : bool isUseRankPort_{false};
321 : bool isUsedRdmaLevel0_{false};
322 : const std::vector<u32>& nicRanksPort_;
323 : const std::vector<u32>& vnicRanksPort_;
324 : bool useSuperPodMode_{false};
325 : const std::vector<HcclIpAddress>& devIpAddr_;
326 : const HcclIpAddress& hostIp_;
327 : const HcclIpAddress& localVnicIp_;
328 : std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap_;
329 : bool devPortSwitchOn_{false};
330 : std::map<u32, TransportType> remoteTransportMap_;
331 :
332 : std::unordered_map<TransportData, LINK> transportMap_;
333 : std::vector<u32> enableP2PDevices_;
334 :
335 : std::vector<std::string> socketTagVec_;
336 : std::vector<DeviceMem> extraMem_;
337 :
338 : bool isGroupMode_ = false;
339 :
340 : std::atomic<bool> stopFlag_{false};
341 : HcclWorkflowMode workflowMode_{HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE};
342 : u64 rankConsistentDataLength_ = 0;
343 : u32 trafficClass_;
344 : u32 serviceLevel_;
345 : u32 ibvCount_ = 0;
346 : std::mutex ibvCountMutex_;
347 : HcclCMDType opType_ = HcclCMDType::HCCL_CMD_INVALID;
348 : bool isStandardCard_ = false;
349 : std::unique_ptr<MulQpInfo> mulQpinfo_ = {nullptr};
350 : std::mutex createSocketMutex_; // BatchSendRecv建链调用CreateDestSockets时，保护socketTagVec_等资源
351 : };
352 : } // namespace hccl
353 :
354 : #endif /* TRANSPORT_MANAGER_H */
|
