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