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 : #include "transport_manager.h"
12 : #include "p2p_mgmt_pub.h"
13 : #include <algorithm>
14 : #include "rank_consistentcy_checker.h"
15 : #include "env_config.h"
16 : #include "detect_connect_anomalies.h"
17 : #include "../../nslbdp/hccl_nslbdp.h"
18 : #include "device_capacity.h"
19 : #include "rt_external.h"
20 :
21 : namespace hccl {
22 :
23 1053 : TransportManager::TransportManager(
24 : CCLBufferManager& cclBufferManager, const std::unique_ptr<HcclSocketManager>& socketManager,
25 : HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool>& notifyPool, const std::vector<RankInfo>& rankInfoList,
26 : RankId userRank, const std::string& identifier, s32 deviceLogicId, NICDeployment nicDeployment, bool isHaveCpuRank,
27 : bool isUseRankPort, bool isUsedRdmaLevel0, const std::vector<u32>& nicRanksPort,
28 : const std::vector<u32>& vnicRanksPort, bool useSuperPodMode, const std::vector<HcclIpAddress>& devIpAddr,
29 1053 : const HcclIpAddress& hostIp, const HcclIpAddress& localVnicIp, std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap)
30 1053 : : cclBufferManager_(cclBufferManager),
31 1053 : socketManager_(socketManager),
32 1053 : dispatcher_(dispatcher),
33 1053 : notifyPool_(notifyPool),
34 1053 : rankInfoList_(rankInfoList),
35 1053 : userRank_(userRank),
36 1053 : identifier_(identifier),
37 1053 : deviceLogicId_(deviceLogicId),
38 1053 : nicDeployment_(nicDeployment),
39 1053 : isHaveCpuRank_(isHaveCpuRank),
40 1053 : isUseRankPort_(isUseRankPort),
41 1053 : isUsedRdmaLevel0_(isUsedRdmaLevel0),
42 1053 : nicRanksPort_(nicRanksPort),
43 1053 : vnicRanksPort_(vnicRanksPort),
44 1053 : useSuperPodMode_(useSuperPodMode),
45 1053 : devIpAddr_(devIpAddr),
46 1053 : hostIp_(hostIp),
47 1053 : localVnicIp_(localVnicIp),
48 1053 : netDevCtxMap_(netDevCtxMap),
49 1053 : trafficClass_(HCCL_COMM_TRAFFIC_CLASS_CONFIG_NOT_SET),
50 2106 : serviceLevel_(HCCL_COMM_SERVICE_LEVEL_CONFIG_NOT_SET)
51 : {
52 1053 : rankConsistentDataLength_ = RankConsistentcyChecker::GetInstance().GetRankConsistentDataLength();
53 1053 : }
54 :
55 1053 : TransportManager::~TransportManager()
56 : {
57 1053 : std::lock_guard<std::mutex> lock(mutex_);
58 1053 : if (enableP2PDevices_.size() != 0) {
59 4 : (void)P2PMgmtPub::DisableP2P(enableP2PDevices_);
60 4 : enableP2PDevices_.clear();
61 : }
62 1051 : }
63 :
64 : constexpr u32 EXCEPTION_DELAY_US_COUNT = 100000;
65 : constexpr u32 MUL_QP_SOCKETS_PER_LINk = 2;
66 0 : HcclResult TransportManager::ExceptionHandle(const std::string& tag, OpCommTransport& opTransportResponse)
67 : {
68 0 : for (auto& levelNSubCommTransport : opTransportResponse) {
69 0 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
70 0 : for (auto& transportRequest : singleSubCommTransport.transportRequests) {
71 0 : if (transportRequest.isValid) {
72 : bool isInterRdma;
73 0 : UpdateIsInterRdma(transportRequest.remoteUserRank, isInterRdma, transportRequest.isUsedRdma);
74 :
75 0 : HcclRankLinkInfo remoteLinkInfo;
76 0 : MakeRemoteLinkInfo(transportRequest.remoteUserRank, isInterRdma, 1, remoteLinkInfo);
77 :
78 0 : HcclIpAddress ipAddr;
79 0 : if (isInterRdma || Is310PDevice()) {
80 0 : ipAddr = nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE ? devIpAddr_[0] : hostIp_;
81 : } else {
82 0 : ipAddr = localVnicIp_;
83 : }
84 :
85 0 : bool isHccs = false;
86 0 : if (!isInterRdma) {
87 0 : isHccs = IsHccsTransport(transportRequest.remoteUserRank, transportRequest.linkType);
88 : }
89 0 : std::string newTag;
90 0 : CHK_RET(ConstructTransTag(tag, newTag, isInterRdma, 0, isHccs));
91 0 : CHK_RET(socketManager_->AddWhiteList(newTag, netDevCtxMap_[ipAddr], remoteLinkInfo));
92 0 : }
93 : }
94 : }
95 : }
96 :
97 0 : return HCCL_SUCCESS;
98 : }
99 :
100 0 : HcclResult TransportManager::CreateVirturalTransport(SingleSubCommTransport& singleSubCommTransport)
101 : {
102 0 : MachinePara machinePara;
103 0 : std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
104 :
105 0 : singleSubCommTransport.virtualLinks.clear();
106 0 : singleSubCommTransport.virtualLinks.resize(singleSubCommTransport.transportRequests.size());
107 :
108 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
109 0 : TransportPara para{};
110 0 : para.virtualFlag = true;
111 0 : para.timeout = kdefaultTimeout;
112 0 : para.index = i;
113 0 : singleSubCommTransport.virtualLinks[i].reset(new (std::nothrow) Transport(
114 0 : TransportType::TRANS_TYPE_RESERVED, para, dispatcher_, notifyPool_, machinePara));
115 0 : CHK_PRT_RET(
116 : !singleSubCommTransport.virtualLinks[i],
117 : HCCL_ERROR("[CreateVirturalTransport]In create link,"
118 : "new link failed"),
119 : HCCL_E_PTR);
120 : }
121 :
122 0 : return HCCL_SUCCESS;
123 0 : }
124 :
125 0 : void TransportManager::SetQpQosAttr(u32 trafficClass, u32 serviceLevel)
126 : {
127 0 : trafficClass_ = trafficClass;
128 0 : serviceLevel_ = serviceLevel;
129 0 : }
130 :
131 0 : HcclResult TransportManager::AddremoteUserRankToList(
132 : TransportRequest& transportRequest, std::vector<u32>& rankList, TransportType transportType)
133 : {
134 0 : if (!transportRequest.isValid) {
135 0 : HCCL_WARNING("[AddremoteUserRankToList]transportRequest is invalid. No need to build a link, skip");
136 0 : return HCCL_SUCCESS;
137 : }
138 0 : TransportType type = TransportType::TRANS_TYPE_RESERVED;
139 0 : CHK_PRT(GetTransportType(transportRequest.remoteUserRank, transportRequest.isUsedRdma, type));
140 0 : if (type == transportType) {
141 : // 仅添加对应Type类型的对端
142 0 : rankList.emplace_back(transportRequest.remoteUserRank);
143 : }
144 0 : return HCCL_SUCCESS;
145 : }
146 :
147 0 : HcclResult TransportManager::GetRemoteRankList(
148 : OpCommTransport& opTransportResponse, std::vector<u32>& rankList, TransportType transportType)
149 : {
150 : // 对当前所有的transportLink做判断
151 0 : for (auto& levelNSubCommTransport : opTransportResponse) {
152 0 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
153 0 : for (auto& transportRequest : singleSubCommTransport.transportRequests) {
154 0 : CHK_PRT(AddremoteUserRankToList(transportRequest, rankList, transportType));
155 : }
156 : }
157 : }
158 0 : return HCCL_SUCCESS;
159 : }
160 :
161 0 : HcclResult TransportManager::createSubCommLinkThreads(
162 : const std::string& tag, const TransportIOMem& transMem, struct SubCommLinkPara& subCommLinkPara, bool isAicpuModeEn,
163 : bool isBackup, u32 subCommIndex, bool isCapture, const HcclCMDType& opType, bool isIndOp, bool isNpuDirectRoce)
164 : {
165 0 : u32 num = subCommLinkPara.remoteRankIdNum;
166 0 : struct SingleSubCommTransport& singleSubCommTransport = subCommLinkPara.singleSubCommTransport;
167 0 : subCommLinkPara.linkThreads.resize(num);
168 0 : subCommLinkPara.linkResult.resize(num, HCCL_SUCCESS);
169 :
170 0 : for (u32 i = 0; i < num; i++) {
171 : u32 index
172 : = subCommLinkPara
173 0 : .remoteRankMap[(subCommLinkPara.remoteRankIdStartIndex + i) % subCommLinkPara.remoteRankMap.size()]
174 0 : .second;
175 0 : auto& transportRequest = singleSubCommTransport.transportRequests[index];
176 0 : auto& link = singleSubCommTransport.links[index];
177 :
178 0 : if ((!transportRequest.isValid) || (link != nullptr) || (isBackup && !transportRequest.isUsedRdma)) {
179 0 : HCCL_INFO(
180 : "[%s]: no need to create p2p back link, remote UserRank[%u], userRank[%u], "
181 : "isUsedRdma[%u], isBackup[%d]",
182 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma, isBackup);
183 0 : continue;
184 : }
185 :
186 0 : DeviceMem inputMem;
187 0 : DeviceMem outputMem;
188 0 : DeviceMem expMem;
189 0 : GetIOMem(transMem, transportRequest.inputMemType, transportRequest.outputMemType, inputMem, outputMem, expMem);
190 0 : HCCL_INFO(
191 : "transportRequest.inputMemType[%d] transportRequest.outputMemType[%d], isBackup[%d]",
192 : transportRequest.inputMemType, transportRequest.outputMemType, isBackup);
193 :
194 0 : IndOpMem indOpMem;
195 0 : if (isIndOp) {
196 0 : indOpMem = transMem.indOpMem;
197 0 : HCCL_DEBUG(
198 : "transportRequest indOpMem, userHostMem size[%llu], userDeviceMem size[%llu]",
199 : indOpMem.userHostMem.size(), indOpMem.userDeviceMem.size());
200 : }
201 :
202 0 : std::vector<std::shared_ptr<HcclSocket>> connectSockets;
203 : bool isInterRdma;
204 0 : bool chooseBackup = transportRequest.isUsedRdma ? isBackup : false;
205 : HcclNetDevCtx netDevCtx;
206 0 : HcclResult ret = CreateDestSockets(
207 : tag, transportRequest.remoteUserRank, singleSubCommTransport.taskNum, connectSockets, netDevCtx,
208 0 : isInterRdma, transportRequest.isUsedRdma, chooseBackup, subCommIndex, transportRequest.linkType);
209 0 : HCCL_DEBUG("[%s]CreateDestSockets finished, chooseBackup[%d]", __func__, chooseBackup);
210 0 : HCCL_DEBUG(
211 : "[%s]: remoteUserRank[%u], userRank[%u], isUsedRdma[%u]", __func__, transportRequest.remoteUserRank,
212 : userRank_, transportRequest.isUsedRdma);
213 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Alloc]Create dest sockets failed"), ret);
214 :
215 0 : MachineType machineType = transportRequest.localUserRank < transportRequest.remoteUserRank ?
216 : MachineType::MACHINE_SERVER_TYPE :
217 : MachineType::MACHINE_CLIENT_TYPE;
218 0 : std::string threadStr = (isInterRdma ? "HcclTerL_" : "HcclIntra_") + std::to_string(i);
219 0 : subCommLinkPara.linkThreads[i].reset(new (std::nothrow) std::thread(
220 0 : &TransportManager::CreateLink, this, tag, hrtErrMGetErrorContextPub(), machineType,
221 0 : rankInfoList_[userRank_].serverId, transportRequest.remoteUserRank,
222 0 : singleSubCommTransport.supportDataReceivedAck, singleSubCommTransport.linkMode,
223 0 : singleSubCommTransport.enableUseOneDoorbell, threadStr, connectSockets, inputMem, outputMem,
224 0 : transportRequest.isUsedRdma, std::ref(link), isAicpuModeEn, std::ref(subCommLinkPara.linkResult[i]),
225 0 : netDevCtx, transportRequest.notifyNum, chooseBackup, isCapture, expMem, transportRequest.linkType, isIndOp,
226 0 : indOpMem, opType, isNpuDirectRoce));
227 0 : CHK_SMART_PTR_NULL(subCommLinkPara.linkThreads[i]); // 异常时其他线程待处理
228 0 : singleSubCommTransport.status[index] = TransportStatus::READY; // 建链后 transport设置为ready状态
229 0 : }
230 :
231 0 : return HCCL_SUCCESS;
232 : }
233 :
234 0 : HcclResult TransportManager::waitSubCommLinkThreadsComplete(struct SubCommLinkPara& subCommLinkPara)
235 : {
236 0 : for (u32 i = 0; i < subCommLinkPara.linkThreads.size(); i++) {
237 0 : if (subCommLinkPara.linkThreads[i] == nullptr || !subCommLinkPara.linkThreads[i]->joinable()) {
238 0 : continue;
239 : }
240 0 : subCommLinkPara.linkThreads[i]->join(); // 等待线程执行完毕
241 0 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
242 : }
243 0 : subCommLinkPara.linkThreads.clear();
244 0 : CHK_PRT_RET(GetStopFlag(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
245 0 : return HCCL_SUCCESS;
246 : }
247 :
248 0 : HcclResult TransportManager::checkSubCommLinkThreadsStatus(
249 : const std::string& tag, struct SubCommLinkPara& subCommLinkPara, bool isBackup)
250 : {
251 0 : u32 num = subCommLinkPara.remoteRankIdNum;
252 0 : struct SingleSubCommTransport& singleSubCommTransport = subCommLinkPara.singleSubCommTransport;
253 :
254 0 : for (u32 i = 0; i < subCommLinkPara.linkResult.size(); i++) {
255 0 : CHK_RET(subCommLinkPara.linkResult[i]);
256 : }
257 0 : for (u32 i = 0; i < num; i++) {
258 : u32 index
259 : = subCommLinkPara
260 0 : .remoteRankMap[(subCommLinkPara.remoteRankIdStartIndex + i) % subCommLinkPara.remoteRankMap.size()]
261 0 : .second;
262 0 : auto& transportRequest = singleSubCommTransport.transportRequests[index];
263 0 : auto& link = singleSubCommTransport.links[index];
264 :
265 0 : if (!transportRequest.isValid) {
266 0 : continue;
267 : }
268 :
269 0 : if (isBackup && !transportRequest.isUsedRdma) {
270 : // 备用链路不需要创建p2p
271 0 : HCCL_INFO(
272 : "[%s]: no need to check p2p backup link, remoteUserRank[%u], userRank[%u], "
273 : "isUsedRdma[%u], isBackup[%d]",
274 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma, isBackup);
275 0 : continue;
276 : }
277 :
278 0 : if (link == nullptr) {
279 0 : HCCL_ERROR(
280 : "[Create]errNo[0x%016llx] transport create fail in thread, local rank[%d] remote rank[%d], "
281 : "inputMemType[%d], outputMemType[%d]",
282 : HCCL_ERROR_CODE(HCCL_E_NOT_FOUND), userRank_, transportRequest.remoteUserRank,
283 : transportRequest.inputMemType, transportRequest.outputMemType);
284 0 : SaluSleep(EXCEPTION_DELAY_US_COUNT);
285 0 : (void)notifyPool_->UnregisterOp(tag);
286 0 : return HCCL_E_NOT_FOUND;
287 : }
288 : }
289 :
290 0 : return HCCL_SUCCESS;
291 : }
292 :
293 0 : HcclResult TransportManager::AllocSubCommLinks(
294 : const std::string& tag, const TransportIOMem& transMem, struct SingleSubCommTransport& singleSubCommTransport,
295 : bool isAicpuModeEn, bool isBackup, u32 subCommIndex, bool isCapture, const HcclCMDType& opType, bool isIndOp,
296 : bool isNpuDirectRoce)
297 : {
298 0 : const u32 offset = 8;
299 0 : std::vector<std::pair<u32, u32>> remoteRankMap;
300 :
301 0 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
302 0 : TransportRequest TemptransportRequest = singleSubCommTransport.transportRequests[i];
303 0 : bool tempIsInterRdma = false;
304 0 : UpdateIsInterRdma(TemptransportRequest.remoteUserRank, tempIsInterRdma, TemptransportRequest.isUsedRdma);
305 0 : if (TemptransportRequest.isValid) {
306 0 : remoteRankMap.push_back(std::make_pair(TemptransportRequest.remoteUserRank, i));
307 0 : if ((rankInfoList_[TemptransportRequest.localUserRank].deviceType == DevType::DEV_TYPE_310P3
308 0 : || isStandardCard_)
309 0 : && !tempIsInterRdma && !Is310PDevice()) {
310 0 : std::vector<u32> enableP2PDevices;
311 0 : enableP2PDevices.push_back(rankInfoList_[TemptransportRequest.remoteUserRank].devicePhyId);
312 0 : HCCL_INFO(
313 : "[Create][DestSockets]localDevicePhyId[%u] enable p2p with remoteDevicePhyId[%u]",
314 : rankInfoList_[TemptransportRequest.localUserRank].devicePhyId,
315 : rankInfoList_[TemptransportRequest.remoteUserRank].devicePhyId);
316 0 : HcclResult ret = P2PMgmtPub::EnableP2P(enableP2PDevices);
317 0 : CHK_PRT_RET(
318 : ret != HCCL_SUCCESS,
319 : HCCL_ERROR(
320 : "[Create][DestSockets]Enable P2P Failed, src devicePhyId[%d], dst devicePhyId[%d], ret[%u]",
321 : rankInfoList_[TemptransportRequest.localUserRank].devicePhyId,
322 : rankInfoList_[TemptransportRequest.remoteUserRank].devicePhyId, ret),
323 : ret);
324 0 : enableP2PDevices_.push_back(rankInfoList_[TemptransportRequest.remoteUserRank].devicePhyId);
325 0 : }
326 : }
327 : }
328 0 : if (remoteRankMap.empty()) {
329 0 : HCCL_INFO("[%s] is empty", __func__);
330 0 : return HCCL_SUCCESS;
331 : }
332 :
333 0 : if (singleSubCommTransport.needVirtualLink) {
334 : // task多线程并行下发，根据当前transport创建vtransport信息
335 0 : CHK_RET(CreateVirturalTransport(singleSubCommTransport));
336 : }
337 :
338 : // sort remoteRankMap by remoteRank
339 : struct LessFirstElement {
340 0 : bool operator()(const std::pair<u32, u32>& a, const std::pair<u32, u32>& b) const { return a.first < b.first; }
341 : };
342 0 : std::sort(remoteRankMap.begin(), remoteRankMap.end(), LessFirstElement());
343 0 : std::vector<std::pair<u32, u32>> reversedRemoteRankMap(remoteRankMap);
344 0 : std::reverse(reversedRemoteRankMap.begin(), reversedRemoteRankMap.end());
345 :
346 0 : struct SubCommLinkPara nextSubCommLinkPara(singleSubCommTransport, remoteRankMap, 0, offset);
347 0 : struct SubCommLinkPara prevSubCommLinkPara(singleSubCommTransport, reversedRemoteRankMap, 0, offset);
348 0 : auto find_greater_than_key1 = [this](const std::pair<u32, u32>& pair) {
349 0 : return pair.first >= (this->userRank_);
350 0 : };
351 0 : auto find_less_than_key1 = [this](const std::pair<u32, u32>& pair) {
352 0 : return pair.first <= (this->userRank_);
353 0 : };
354 0 : auto nextIt = find_if(remoteRankMap.begin(), remoteRankMap.end(), find_greater_than_key1);
355 0 : auto prevIt = find_if(reversedRemoteRankMap.begin(), reversedRemoteRankMap.end(), find_less_than_key1);
356 0 : u32 rankNum = remoteRankMap.size();
357 0 : nextSubCommLinkPara.remoteRankIdStartIndex = std::distance(remoteRankMap.begin(), nextIt) % rankNum;
358 0 : prevSubCommLinkPara.remoteRankIdStartIndex = std::distance(reversedRemoteRankMap.begin(), prevIt) % rankNum;
359 :
360 0 : for (u32 i = 0; i < (rankNum / (FACTOR_NUM_TWO * offset)) + 1; i++) {
361 0 : if ((i == rankNum / (FACTOR_NUM_TWO * offset)) && (rankNum % (FACTOR_NUM_TWO * offset)) != 0) {
362 0 : nextSubCommLinkPara.remoteRankIdNum = (rankNum % (FACTOR_NUM_TWO * offset)) / FACTOR_NUM_TWO
363 0 : + ((rankNum % (FACTOR_NUM_TWO * offset)) % FACTOR_NUM_TWO);
364 0 : prevSubCommLinkPara.remoteRankIdNum = (rankNum % (FACTOR_NUM_TWO * offset)) / FACTOR_NUM_TWO;
365 : }
366 :
367 0 : CHK_RET(createSubCommLinkThreads(
368 : tag, transMem, nextSubCommLinkPara, isAicpuModeEn, isBackup, subCommIndex, isCapture, opType, isIndOp,
369 : isNpuDirectRoce));
370 0 : CHK_RET(createSubCommLinkThreads(
371 : tag, transMem, prevSubCommLinkPara, isAicpuModeEn, isBackup, subCommIndex, isCapture, opType, isIndOp,
372 : isNpuDirectRoce));
373 0 : CHK_RET(waitSubCommLinkThreadsComplete(nextSubCommLinkPara));
374 0 : CHK_RET(waitSubCommLinkThreadsComplete(prevSubCommLinkPara));
375 0 : CHK_RET(checkSubCommLinkThreadsStatus(tag, nextSubCommLinkPara, isBackup));
376 0 : CHK_RET(checkSubCommLinkThreadsStatus(tag, prevSubCommLinkPara, isBackup));
377 0 : for (auto& tmpTag : socketTagVec_) {
378 0 : (void)socketManager_->DestroySockets(tmpTag);
379 : }
380 0 : socketTagVec_.clear();
381 :
382 0 : nextSubCommLinkPara.remoteRankIdStartIndex += offset;
383 0 : prevSubCommLinkPara.remoteRankIdStartIndex += offset;
384 : }
385 :
386 0 : return HCCL_SUCCESS;
387 0 : }
388 :
389 6 : HcclResult TransportManager::CreateBatchSendRecvLinks(
390 : const std::string& tag, const TransportIOMem& transMem, struct LinkPoolPara& linkPoolPara, bool isAicpuModeEn,
391 : bool isBackup, u32 subCommIndex, bool isCapture, const HcclCMDType& opType, bool isIndOp)
392 : {
393 6 : HcclResult ret = hrtSetDevice(deviceLogicId_);
394 6 : if (ret != HCCL_SUCCESS) {
395 2 : HCCL_ERROR("[CreateBatchSendRecvLinks]hrtSetDevice failed, ret[%d]", ret);
396 2 : linkPoolPara.abortFlag = true;
397 2 : return ret;
398 : }
399 4 : struct SingleSubCommTransport& singleSubCommTransport = linkPoolPara.singleSubCommTransport;
400 4 : u32 currentIdx = 0;
401 4 : u32 requestIdx = 0;
402 : while (true) {
403 6 : currentIdx = linkPoolPara.taskIndex.fetch_add(1);
404 6 : if (currentIdx >= linkPoolPara.taskList.size() || linkPoolPara.abortFlag) {
405 2 : break;
406 : }
407 4 : requestIdx = linkPoolPara.taskList[currentIdx].second;
408 :
409 4 : auto& transportRequest = singleSubCommTransport.transportRequests[requestIdx];
410 4 : auto& link = singleSubCommTransport.links[requestIdx];
411 :
412 : // 无效请求、link已创建、备用链路，这三种情况不需要创建link
413 4 : if ((!transportRequest.isValid) || (link != nullptr) || (isBackup && !transportRequest.isUsedRdma)) {
414 0 : HCCL_INFO(
415 : "[%s]: no need to create p2p back link, remote UserRank[%u], userRank[%u], "
416 : "isUsedRdma[%u], isBackup[%d]",
417 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma, isBackup);
418 0 : continue;
419 : }
420 :
421 4 : DeviceMem inputMem;
422 4 : DeviceMem outputMem;
423 4 : DeviceMem expMem;
424 4 : GetIOMem(transMem, transportRequest.inputMemType, transportRequest.outputMemType, inputMem, outputMem, expMem);
425 4 : HCCL_INFO(
426 : "[CreateBatchSendRecvLinks]transportRequest.inputMemType[%d] transportRequest.outputMemType[%d], "
427 : "isBackup[%d]",
428 : transportRequest.inputMemType, transportRequest.outputMemType, isBackup);
429 :
430 4 : IndOpMem indOpMem;
431 4 : if (isIndOp) {
432 0 : indOpMem = transMem.indOpMem;
433 0 : HCCL_DEBUG(
434 : "[CreateBatchSendRecvLinks]transportRequest indOpMem, userHostMem size[%llu], userDeviceMem size[%llu]",
435 : indOpMem.userHostMem.size(), indOpMem.userDeviceMem.size());
436 : }
437 :
438 4 : std::vector<std::shared_ptr<HcclSocket>> connectSockets;
439 4 : bool isInterRdma = false;
440 4 : bool chooseBackup = transportRequest.isUsedRdma ? isBackup : false;
441 : HcclNetDevCtx netDevCtx;
442 : {
443 4 : std::lock_guard<std::mutex> lock(createSocketMutex_);
444 8 : ret = CreateDestSockets(
445 : tag, transportRequest.remoteUserRank, singleSubCommTransport.taskNum, connectSockets, netDevCtx,
446 4 : isInterRdma, transportRequest.isUsedRdma, chooseBackup, subCommIndex, transportRequest.linkType);
447 4 : }
448 4 : HCCL_DEBUG(
449 : "[%s]CreateDestSockets finished, chooseBackup[%d], remoteUserRank[%u], userRank[%u], isUsedRdma[%u]",
450 : __func__, chooseBackup, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma);
451 4 : if (ret != HCCL_SUCCESS) {
452 2 : HCCL_ERROR("[CreateBatchSendRecvLinks]Create dest sockets failed");
453 2 : linkPoolPara.linkResults[currentIdx] = ret;
454 2 : linkPoolPara.abortFlag = true;
455 2 : return ret;
456 : }
457 :
458 2 : MachineType machineType = transportRequest.localUserRank < transportRequest.remoteUserRank ?
459 : MachineType::MACHINE_SERVER_TYPE :
460 : MachineType::MACHINE_CLIENT_TYPE;
461 2 : std::string threadStr = (isInterRdma ? "HcclTerL_" : "HcclIntra_") + std::to_string(requestIdx);
462 2 : HCCL_INFO("[%s]threadStr[%s], poolName[%s]", __func__, threadStr.c_str(), linkPoolPara.poolName.c_str());
463 8 : ret = CreateLink(
464 4 : tag, hrtErrMGetErrorContextPub(), machineType, rankInfoList_[userRank_].serverId,
465 2 : transportRequest.remoteUserRank, singleSubCommTransport.supportDataReceivedAck,
466 2 : singleSubCommTransport.linkMode, singleSubCommTransport.enableUseOneDoorbell, threadStr, connectSockets,
467 2 : inputMem, outputMem, transportRequest.isUsedRdma, link, isAicpuModeEn, linkPoolPara.linkResults[currentIdx],
468 : netDevCtx, transportRequest.notifyNum, chooseBackup, isCapture, expMem, transportRequest.linkType, isIndOp,
469 : indOpMem, opType, false);
470 2 : if (ret != HCCL_SUCCESS) {
471 0 : HCCL_ERROR("[CreateBatchSendRecvLinks]Create Link failed");
472 0 : linkPoolPara.linkResults[currentIdx] = ret;
473 0 : (void)hrtResetDevice(deviceLogicId_); // CreateLink会调用一次hrtSetDevice
474 0 : linkPoolPara.abortFlag = true;
475 0 : return ret;
476 : }
477 2 : ret = hrtResetDevice(deviceLogicId_);
478 2 : if (ret != HCCL_SUCCESS) {
479 0 : HCCL_ERROR("[CreateBatchSendRecvLinks]hrtResetDevice failed");
480 0 : linkPoolPara.linkResults[currentIdx] = ret;
481 0 : linkPoolPara.abortFlag = true;
482 0 : return ret;
483 : }
484 2 : singleSubCommTransport.status[requestIdx] = TransportStatus::READY; // 建链后 transport设置为ready状态
485 14 : }
486 2 : return HCCL_SUCCESS;
487 : }
488 :
489 4 : HcclResult TransportManager::WaitBatchSendRecvThreadsComplete(struct LinkPoolPara& linkPoolPara)
490 : {
491 8 : for (u32 i = 0; i < linkPoolPara.linkThreads.size(); i++) {
492 4 : if (linkPoolPara.linkThreads[i] == nullptr || !linkPoolPara.linkThreads[i]->joinable()) {
493 0 : continue;
494 : }
495 4 : linkPoolPara.linkThreads[i]->join(); // 等待线程执行完毕
496 4 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
497 : }
498 4 : linkPoolPara.linkThreads.clear();
499 4 : CHK_PRT_RET(GetStopFlag(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
500 :
501 7 : for (u32 i = 0; i < linkPoolPara.linkResults.size(); i++) {
502 4 : CHK_RET(linkPoolPara.linkResults[i]);
503 : }
504 3 : CHK_PRT_RET(
505 : linkPoolPara.abortFlag, HCCL_ERROR("[WaitBatchSendRecvThreadsComplete] abortFlag is set"), HCCL_E_INTERNAL);
506 :
507 2 : return HCCL_SUCCESS;
508 : }
509 :
510 1 : HcclResult TransportManager::CheckBatchSendRecvLinkStatus(
511 : const std::string& tag, struct SingleSubCommTransport& singleSubCommTransport, bool isBackup)
512 : {
513 1 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); ++i) {
514 1 : auto& transportRequest = singleSubCommTransport.transportRequests[i];
515 1 : if (transportRequest.isValid) {
516 : // 备用链路不需要创建p2p
517 1 : if (isBackup && !transportRequest.isUsedRdma) {
518 0 : HCCL_INFO(
519 : "[%s]: no need to check p2p backup link, remoteUserRank[%u], userRank[%u], "
520 : "isUsedRdma[%u], isBackup[%d]",
521 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma, isBackup);
522 0 : continue;
523 : }
524 :
525 1 : if (singleSubCommTransport.links[i] == nullptr) {
526 1 : HCCL_ERROR(
527 : "[Create]errNo[0x%016llx] transport create fail in thread, local rank[%u] remote rank[%u], "
528 : "inputMemType[%d], outputMemType[%d]",
529 : HCCL_ERROR_CODE(HCCL_E_NOT_FOUND), userRank_, transportRequest.remoteUserRank,
530 : transportRequest.inputMemType, transportRequest.outputMemType);
531 1 : SaluSleep(EXCEPTION_DELAY_US_COUNT);
532 1 : (void)notifyPool_->UnregisterOp(tag);
533 1 : return HCCL_E_NOT_FOUND;
534 : }
535 : }
536 : }
537 :
538 0 : for (auto& tmpTag : socketTagVec_) {
539 0 : (void)socketManager_->DestroySockets(tmpTag);
540 : }
541 0 : socketTagVec_.clear();
542 :
543 0 : return HCCL_SUCCESS;
544 : }
545 :
546 3 : HcclResult TransportManager::PrepareTaskLists(
547 : HcclSendRecvItem* sendRecvItemsPtr, u32 itemNum, const SingleSubCommTransport& singleSubCommTransport,
548 : std::vector<std::pair<u32, u32>>& senderList, std::vector<std::pair<u32, u32>>& receiverList)
549 : {
550 3 : if (sendRecvItemsPtr == nullptr || itemNum == 0) {
551 0 : HCCL_INFO("[%s] SendRecvItemsPtr is empty", __func__);
552 0 : return HCCL_SUCCESS;
553 : }
554 :
555 3 : std::unordered_set<u32> senderSet;
556 3 : std::unordered_set<u32> receiverSet;
557 :
558 12 : for (u32 i = 0; i < itemNum; ++i) {
559 9 : if (sendRecvItemsPtr[i].sendRecvType == HcclSendRecvType::HCCL_SEND) {
560 3 : receiverSet.insert(sendRecvItemsPtr[i].remoteRank);
561 6 : } else if (sendRecvItemsPtr[i].sendRecvType == HcclSendRecvType::HCCL_RECV) {
562 6 : senderSet.insert(sendRecvItemsPtr[i].remoteRank);
563 : }
564 : }
565 :
566 9 : for (u32 i = 0; i < singleSubCommTransport.transportRequests.size(); i++) {
567 6 : if (singleSubCommTransport.transportRequests[i].isValid) {
568 6 : u32 remoteRank = singleSubCommTransport.transportRequests[i].remoteUserRank;
569 6 : bool isSender = senderSet.count(remoteRank);
570 6 : bool isReceiver = receiverSet.count(remoteRank);
571 6 : if (isSender && (!isReceiver || remoteRank < userRank_)) {
572 3 : senderList.emplace_back(std::make_pair(remoteRank, i));
573 3 : } else if (isReceiver) {
574 3 : receiverList.emplace_back(std::make_pair(remoteRank, i));
575 : }
576 : }
577 : }
578 :
579 0 : auto cmp = [](const std::pair<u32, u32>& a, const std::pair<u32, u32>& b) {
580 0 : return a.first < b.first;
581 : };
582 3 : std::sort(senderList.begin(), senderList.end(), cmp);
583 3 : std::sort(receiverList.begin(), receiverList.end(), cmp);
584 :
585 3 : return HCCL_SUCCESS;
586 3 : }
587 :
588 3 : HcclResult TransportManager::AllocBatchSendRecvLinks(
589 : HcclSendRecvItem* sendRecvItemsPtr, u32 itemNum, const std::string& tag, const TransportIOMem& transMem,
590 : struct SingleSubCommTransport& singleSubCommTransport, bool isAicpuModeEn, bool isBackup, u32 subCommIndex,
591 : bool isCapture, const HcclCMDType& opType, bool isIndOp)
592 : {
593 : // 记录pair<remoteRank, idx>, idx表示remoteRank对应的建链信息在transportRequests中的索引位置
594 3 : std::vector<std::pair<u32, u32>> senderList;
595 3 : std::vector<std::pair<u32, u32>> receiverList;
596 :
597 3 : CHK_RET(PrepareTaskLists(sendRecvItemsPtr, itemNum, singleSubCommTransport, senderList, receiverList));
598 3 : if (senderList.empty() && receiverList.empty()) {
599 0 : HCCL_INFO("[%s] TransportRequests is empty", __func__);
600 0 : return HCCL_SUCCESS;
601 : }
602 :
603 3 : if (singleSubCommTransport.needVirtualLink) {
604 : // task多线程并行下发，根据当前transport创建vtransport信息
605 0 : CHK_RET(CreateVirturalTransport(singleSubCommTransport));
606 : }
607 :
608 6 : struct LinkPoolPara senderLinkPoolPara(singleSubCommTransport, "sender", senderList);
609 3 : struct LinkPoolPara receiverLinkPoolPara(singleSubCommTransport, "receiver", receiverList);
610 6 : for (u32 i = 0; i < senderLinkPoolPara.linkThreads.size(); ++i) {
611 6 : senderLinkPoolPara.linkThreads[i].reset(new (std::nothrow) std::thread(
612 3 : &TransportManager::CreateBatchSendRecvLinks, this, tag, std::ref(transMem), std::ref(senderLinkPoolPara),
613 6 : isAicpuModeEn, isBackup, subCommIndex, isCapture, opType, isIndOp));
614 :
615 3 : if (senderLinkPoolPara.linkThreads[i] == nullptr) {
616 0 : HCCL_ERROR("[AllocBatchSendRecvLinks] Failed to create sender thread %u", i);
617 0 : senderLinkPoolPara.abortFlag = true;
618 0 : WaitBatchSendRecvThreadsComplete(senderLinkPoolPara); // 清理已建立的线程
619 0 : return HCCL_E_PTR;
620 : }
621 : }
622 6 : for (u32 i = 0; i < receiverLinkPoolPara.linkThreads.size(); ++i) {
623 6 : receiverLinkPoolPara.linkThreads[i].reset(new (std::nothrow) std::thread(
624 3 : &TransportManager::CreateBatchSendRecvLinks, this, tag, std::ref(transMem), std::ref(receiverLinkPoolPara),
625 6 : isAicpuModeEn, isBackup, subCommIndex, isCapture, opType, isIndOp));
626 :
627 3 : if (receiverLinkPoolPara.linkThreads[i] == nullptr) {
628 0 : HCCL_ERROR("[AllocBatchSendRecvLinks] Failed to create receiver thread %u", i);
629 0 : receiverLinkPoolPara.abortFlag = true;
630 0 : senderLinkPoolPara.abortFlag = true;
631 0 : WaitBatchSendRecvThreadsComplete(senderLinkPoolPara);
632 0 : WaitBatchSendRecvThreadsComplete(receiverLinkPoolPara);
633 0 : return HCCL_E_PTR;
634 : }
635 : }
636 :
637 3 : CHK_RET(WaitBatchSendRecvThreadsComplete(senderLinkPoolPara));
638 1 : CHK_RET(WaitBatchSendRecvThreadsComplete(receiverLinkPoolPara));
639 1 : CHK_RET(CheckBatchSendRecvLinkStatus(tag, singleSubCommTransport, isBackup));
640 :
641 1 : return HCCL_SUCCESS;
642 3 : }
643 :
644 52 : HcclResult TransportManager::Alloc(
645 : const std::string& tag, const TransportIOMem& transMem, OpCommTransport& opTransportResponse, bool isAicpuModeEn,
646 : bool isBackup, bool isZeroCopy, const HcclCMDType& opType, bool isCapture, bool isIndOp, bool isNpuDirectRoce,
647 : const OpParam* opParam)
648 : {
649 52 : std::lock_guard<std::mutex> lock(mutex_);
650 52 : CHK_RET(notifyPool_->RegisterOp(tag));
651 52 : workflowMode_ = GetWorkflowMode(); // 后续有起新的线程，因此更新一下workflowMode
652 52 : for (u32 levelIdx = 0; levelIdx < opTransportResponse.size(); levelIdx++) {
653 52 : auto& levelNSubCommTransport = opTransportResponse[levelIdx];
654 52 : u32 subCommIndex = 0;
655 52 : for (auto& singleSubCommTransport : levelNSubCommTransport) {
656 52 : subCommIndex++;
657 : DevType devType;
658 102 : CHK_RET(hrtGetDeviceType(devType));
659 52 : if (devType == DevType::DEV_TYPE_910_93) {
660 : // 如果是零拷贝场景下level0通信域交换零拷贝的共享内存
661 0 : if (levelIdx == COMM_LEVEL0 && isZeroCopy) {
662 0 : for (auto& transportRequest : singleSubCommTransport.transportRequests) {
663 0 : if (transportRequest.inputMemType != TransportMemType::RESERVED) {
664 0 : transportRequest.inputMemType = TransportMemType::PARAM_INPUT;
665 : }
666 0 : if (transportRequest.outputMemType != TransportMemType::RESERVED) {
667 0 : transportRequest.outputMemType = (opType == HcclCMDType::HCCL_CMD_BROADCAST) ?
668 : TransportMemType::PARAM_INPUT :
669 : TransportMemType::PARAM_OUTPUT;
670 : }
671 : }
672 : }
673 0 : if (opType == HcclCMDType::HCCL_CMD_BATCH_SEND_RECV) {
674 0 : CHK_PTR_NULL(opParam);
675 0 : CHK_RET(AllocBatchSendRecvLinks(
676 : opParam->BatchSendRecvDataDes.sendRecvItemsPtr, opParam->BatchSendRecvDataDes.itemNum, tag,
677 : transMem, singleSubCommTransport, isAicpuModeEn, isBackup, subCommIndex, isCapture, opType,
678 : isIndOp));
679 : } else {
680 0 : CHK_RET(AllocSubCommLinks(
681 : tag, transMem, singleSubCommTransport, isAicpuModeEn, isBackup, subCommIndex, isCapture, opType,
682 : isIndOp, isNpuDirectRoce));
683 : }
684 0 : continue;
685 0 : }
686 :
687 52 : std::vector<std::unique_ptr<std::thread>> linkThreads; // 建链所需线程
688 52 : std::vector<HcclResult> linkResult; // CreateLink返回值出参
689 52 : linkThreads.resize(singleSubCommTransport.transportRequests.size());
690 52 : linkResult.resize(linkThreads.size(), HCCL_SUCCESS);
691 50 : ThreadsGuard threadsGuard(linkThreads); // 确保异常退出场景析构时等待线程join
692 48 : u32 threadsRapplyNum{0}; // 线程使用计数器
693 :
694 48 : if (singleSubCommTransport.needVirtualLink) {
695 : // task多线程并行下发，根据当前transport创建vtransport信息
696 0 : CHK_RET(CreateVirturalTransport(singleSubCommTransport));
697 : }
698 :
699 48 : u32 linkIdx = 0;
700 100 : for (auto& transportRequest : singleSubCommTransport.transportRequests) {
701 87 : if (transportRequest.isValid && singleSubCommTransport.links[linkIdx] == nullptr) {
702 73 : if (isBackup && !transportRequest.isUsedRdma) {
703 : // 备用链路不需要创建p2p
704 0 : HCCL_INFO(
705 : "[%s]: no need to create p2p backup link, remoteUserRank[%u], userRank[%u], "
706 : "isUsedRdma[%u], isBackup[%d]",
707 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma,
708 : isBackup);
709 0 : linkIdx++;
710 0 : continue;
711 : }
712 73 : bool tempIsInterRdma = false;
713 73 : UpdateIsInterRdma(transportRequest.remoteUserRank, tempIsInterRdma, transportRequest.isUsedRdma);
714 77 : if ((rankInfoList_[transportRequest.localUserRank].deviceType == DevType::DEV_TYPE_310P3
715 72 : || isStandardCard_)
716 149 : && !tempIsInterRdma && !Is310PDevice()) {
717 5 : std::vector<u32> enableP2PDevices;
718 5 : enableP2PDevices.push_back(rankInfoList_[transportRequest.remoteUserRank].devicePhyId);
719 5 : HCCL_INFO(
720 : "[Alloc]localDevicePhyId[%u] enable p2p with remoteDevicePhyId[%u]",
721 : rankInfoList_[transportRequest.localUserRank].devicePhyId,
722 : rankInfoList_[transportRequest.remoteUserRank].devicePhyId);
723 5 : HcclResult ret = P2PMgmtPub::EnableP2P(enableP2PDevices);
724 5 : CHK_PRT_RET(
725 : ret != HCCL_SUCCESS,
726 : HCCL_ERROR(
727 : "[Alloc]Enable P2P Failed, src devicePhyId[%d], dst devicePhyId[%d], ret[%u]",
728 : rankInfoList_[transportRequest.localUserRank].devicePhyId,
729 : rankInfoList_[transportRequest.remoteUserRank].devicePhyId, ret),
730 : ret);
731 5 : enableP2PDevices_.push_back(rankInfoList_[transportRequest.remoteUserRank].devicePhyId);
732 5 : }
733 77 : DeviceMem inputMem;
734 77 : DeviceMem outputMem;
735 77 : DeviceMem expMem;
736 77 : HCCL_INFO(
737 : "transportRequest.inputMemType[%d] transportRequest.outputMemType[%d], isBackup[%d]",
738 : transportRequest.inputMemType, transportRequest.outputMemType, isBackup);
739 77 : GetIOMem(
740 : transMem, transportRequest.inputMemType, transportRequest.outputMemType, inputMem, outputMem,
741 : expMem);
742 :
743 76 : IndOpMem indOpMem;
744 76 : if (isIndOp) {
745 0 : indOpMem = transMem.indOpMem;
746 0 : HCCL_DEBUG(
747 : "transportRequest indOpMem, userHostMem size[%llu], userDeviceMem size[%llu]",
748 : indOpMem.userHostMem.size(), indOpMem.userDeviceMem.size());
749 : }
750 :
751 76 : std::vector<std::shared_ptr<HcclSocket>> connectSockets;
752 : bool isInterRdma;
753 77 : HCCL_DEBUG(
754 : "[%s]: remoteUserRank[%u], userRank[%u], isUsedRdma[%u], tag[%s]", __func__,
755 : transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma, tag.c_str());
756 77 : bool chooseBackup = transportRequest.isUsedRdma ? isBackup : false;
757 : HcclNetDevCtx netDevCtx;
758 154 : HcclResult ret = CreateDestSockets(
759 : tag, transportRequest.remoteUserRank, singleSubCommTransport.taskNum, connectSockets, netDevCtx,
760 77 : isInterRdma, transportRequest.isUsedRdma, chooseBackup, subCommIndex,
761 : transportRequest.linkType);
762 77 : HCCL_DEBUG("[%s]CreateDestSockets finished, chooseBackup[%d]", __func__, chooseBackup);
763 77 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Alloc]Create dest sockets failed"), ret);
764 :
765 38 : MachineType machineType = transportRequest.localUserRank < transportRequest.remoteUserRank ?
766 : MachineType::MACHINE_SERVER_TYPE :
767 : MachineType::MACHINE_CLIENT_TYPE;
768 :
769 38 : if (transportRequest.isUsedRdma) {
770 0 : HCCL_INFO(
771 : "[%s]: create rdma link, remoteUserRank[%u], userRank[%u], "
772 : "isBackup[%d], chooseBackup[%d], isInterRdma[%d]",
773 : __func__, transportRequest.remoteUserRank, userRank_, isBackup, chooseBackup, isInterRdma);
774 : }
775 38 : bool chooseAivRoceDirect = transportRequest.isUsedRdma ? isNpuDirectRoce : false;
776 : std::string threadStr
777 38 : = (isInterRdma ? "HcclTerL_" : "HcclIntra_") + std::to_string(threadsRapplyNum);
778 76 : linkThreads[threadsRapplyNum].reset(new (std::nothrow) std::thread(
779 0 : &TransportManager::CreateLink, this, tag, hrtErrMGetErrorContextPub(), machineType,
780 38 : rankInfoList_[userRank_].serverId, transportRequest.remoteUserRank,
781 38 : singleSubCommTransport.supportDataReceivedAck, singleSubCommTransport.linkMode,
782 38 : singleSubCommTransport.enableUseOneDoorbell, threadStr, connectSockets, inputMem, outputMem,
783 38 : transportRequest.isUsedRdma, std::ref(singleSubCommTransport.links[linkIdx]), isAicpuModeEn,
784 38 : std::ref(linkResult[threadsRapplyNum]), netDevCtx, transportRequest.notifyNum, chooseBackup,
785 76 : isCapture, expMem, transportRequest.linkType, isIndOp, indOpMem, opType, chooseAivRoceDirect));
786 38 : CHK_SMART_PTR_NULL(linkThreads[threadsRapplyNum]); // 异常时其他线程待处理
787 38 : singleSubCommTransport.status[linkIdx] = TransportStatus::READY; // 建链后 transport设置为ready状态
788 38 : threadsRapplyNum++;
789 233 : }
790 52 : linkIdx++;
791 : }
792 :
793 65 : for (u32 index = 0; index < linkThreads.size(); index++) {
794 52 : if (linkThreads[index] == nullptr || !linkThreads[index]->joinable()) {
795 14 : continue;
796 : }
797 38 : linkThreads[index]->join(); // 等待线程执行完毕
798 38 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
799 : }
800 13 : linkThreads.clear();
801 13 : CHK_PRT_RET(GetStopFlag(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
802 13 : for (u32 index = 0; index < linkResult.size(); index++) {
803 13 : CHK_RET(linkResult[index]);
804 : }
805 :
806 0 : linkIdx = 0;
807 0 : for (auto& transportRequest : singleSubCommTransport.transportRequests) {
808 0 : if (transportRequest.isValid) {
809 0 : if (isBackup && !transportRequest.isUsedRdma) {
810 : // 备用链路不需要创建p2p
811 0 : HCCL_INFO(
812 : "[%s]: no need to check p2p backup link, remoteUserRank[%u], userRank[%u], "
813 : "isUsedRdma[%u], isBackup[%d]",
814 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma,
815 : isBackup);
816 0 : linkIdx++;
817 0 : continue;
818 : }
819 0 : if (singleSubCommTransport.links[linkIdx] == nullptr) {
820 0 : HCCL_ERROR(
821 : "[Create]errNo[0x%016llx] transport create fail in thread, local rank[%d] remote rank[%d]",
822 : HCCL_ERROR_CODE(HCCL_E_NOT_FOUND), userRank_, transportRequest.remoteUserRank);
823 0 : (void)ExceptionHandle(tag, opTransportResponse);
824 0 : SaluSleep(EXCEPTION_DELAY_US_COUNT);
825 0 : (void)notifyPool_->UnregisterOp(tag);
826 0 : return HCCL_E_NOT_FOUND;
827 : }
828 : }
829 0 : linkIdx++;
830 : }
831 0 : for (auto& tmpTag : socketTagVec_) {
832 0 : (void)socketManager_->DestroySockets(tmpTag);
833 : }
834 0 : socketTagVec_.clear();
835 155 : }
836 : }
837 0 : CHK_RET(notifyPool_->UnregisterOp(tag));
838 0 : return HCCL_SUCCESS;
839 50 : }
840 :
841 0 : HcclResult TransportManager::GetIncreRemoteRankList(
842 : OpCommTransport& opTransportReq, std::vector<u32>& rankList, TransportType transportType)
843 : {
844 0 : for (u32 levelIndex = 0; levelIndex < opTransportReq.size(); levelIndex++) {
845 0 : for (u32 ringIndex = 0; ringIndex < opTransportReq[levelIndex].size(); ringIndex++) {
846 0 : SingleSubCommTransport& reqSingleSubComm = opTransportReq[levelIndex][ringIndex];
847 0 : for (u32 rankIndex = 0; rankIndex < reqSingleSubComm.transportRequests.size(); rankIndex++) {
848 0 : TransportRequest& transportRequest = reqSingleSubComm.transportRequests[rankIndex];
849 0 : CHK_PRT(AddremoteUserRankToList(transportRequest, rankList, transportType));
850 : }
851 : }
852 : }
853 0 : return HCCL_SUCCESS;
854 : }
855 :
856 0 : HcclResult TransportManager::IncreAlloc(
857 : const std::string& tag, const TransportIOMem& transMem, OpCommTransport& opTransportReq,
858 : OpCommTransport& opTransportResponse, bool isAicpuModeEn, bool isBackup, bool isCapture, const HcclCMDType& opType)
859 : {
860 0 : std::lock_guard<std::mutex> lock(mutex_);
861 0 : CHK_RET(notifyPool_->RegisterOp(tag));
862 :
863 0 : workflowMode_ = GetWorkflowMode();
864 0 : for (u32 levelIndex = 0; levelIndex < opTransportReq.size(); levelIndex++) {
865 0 : u32 subCommIndex = 0;
866 0 : for (u32 ringIndex = 0; ringIndex < opTransportReq[levelIndex].size(); ringIndex++) {
867 0 : subCommIndex++;
868 0 : std::vector<std::unique_ptr<std::thread>> linkThreads; // 建链所需线程
869 0 : std::vector<HcclResult> linkResult; // CreateLink返回值出参
870 0 : linkThreads.resize(opTransportReq[levelIndex][ringIndex].transportRequests.size());
871 0 : linkResult.resize(linkThreads.size(), HCCL_SUCCESS);
872 0 : ThreadsGuard threadsGuard(linkThreads); // 确保异常退出场景析构时等待线程join
873 0 : u32 threadsRapplyNum{0}; // 线程使用计数器
874 0 : SingleSubCommTransport& reqSingleSubComm = opTransportReq[levelIndex][ringIndex];
875 0 : SingleSubCommTransport& respSingleSubComm = opTransportResponse[levelIndex][ringIndex];
876 0 : for (u32 rankIndex = 0; rankIndex < reqSingleSubComm.transportRequests.size(); rankIndex++) {
877 0 : TransportRequest& transportRequest = reqSingleSubComm.transportRequests[rankIndex];
878 0 : CHK_PRT_RET(
879 : rankIndex >= respSingleSubComm.links.size(),
880 : HCCL_ERROR(
881 : "[IncreAlloc] The remote rank_id[%u] is larger than the existent respSingleSubComm map "
882 : "size[%u]",
883 : rankIndex, respSingleSubComm.links.size()),
884 : HCCL_E_PARA);
885 0 : if (respSingleSubComm.links[rankIndex] != nullptr
886 0 : && respSingleSubComm.links[rankIndex]->GetLinkType() != hccl::LinkType::LINK_RESERVED) {
887 0 : HCCL_INFO(
888 : "[IncreAlloc] The link to remote userRank[%u] has existed", transportRequest.remoteUserRank);
889 0 : continue;
890 : }
891 0 : if (transportRequest.isValid) {
892 0 : if (isBackup && !transportRequest.isUsedRdma) {
893 : // 备用链路不需要创建p2p
894 0 : HCCL_INFO(
895 : "[%s]: no need to create p2p backup link, remoteUserRank[%u], userRank[%u], "
896 : "isUsedRdma[%u], isBackup[%d]",
897 : __func__, transportRequest.remoteUserRank, userRank_, transportRequest.isUsedRdma,
898 : isBackup);
899 0 : continue;
900 : }
901 0 : bool tempIsInterRdma = false;
902 0 : UpdateIsInterRdma(transportRequest.remoteUserRank, tempIsInterRdma, transportRequest.isUsedRdma);
903 0 : if ((rankInfoList_[transportRequest.localUserRank].deviceType == DevType::DEV_TYPE_310P3
904 0 : || isStandardCard_)
905 0 : && !tempIsInterRdma && !Is310PDevice()) {
906 0 : std::vector<u32> enableP2PDevices;
907 0 : enableP2PDevices.push_back(rankInfoList_[transportRequest.remoteUserRank].devicePhyId);
908 0 : HCCL_INFO(
909 : "[IncreAlloc]localDevicePhyId[%u] enable p2p with remoteDevicePhyId[%u]",
910 : rankInfoList_[transportRequest.localUserRank].devicePhyId,
911 : rankInfoList_[transportRequest.remoteUserRank].devicePhyId);
912 0 : HcclResult ret = P2PMgmtPub::EnableP2P(enableP2PDevices);
913 0 : CHK_PRT_RET(
914 : ret != HCCL_SUCCESS,
915 : HCCL_ERROR(
916 : "[IncreAlloc]Enable P2P Failed, src devicePhyId[%d], dst devicePhyId[%d], ret[%u]",
917 : rankInfoList_[transportRequest.localUserRank].devicePhyId,
918 : rankInfoList_[transportRequest.remoteUserRank].devicePhyId, ret),
919 : ret);
920 0 : enableP2PDevices_.push_back(rankInfoList_[transportRequest.remoteUserRank].devicePhyId);
921 0 : }
922 0 : respSingleSubComm.transportRequests[rankIndex] = transportRequest;
923 0 : DeviceMem inputMem;
924 0 : DeviceMem outputMem;
925 0 : DeviceMem expMem;
926 0 : GetIOMem(
927 : transMem, transportRequest.inputMemType, transportRequest.outputMemType, inputMem, outputMem,
928 : expMem);
929 0 : HCCL_INFO(
930 : "[IncreAlloc] transportRequest.inputMemType[%d] transportRequest.outputMemType[%d], "
931 : "isBackup[%d]",
932 : transportRequest.inputMemType, transportRequest.outputMemType, isBackup);
933 :
934 0 : std::vector<std::shared_ptr<HcclSocket>> connectSockets;
935 : bool isInterRdma;
936 0 : bool chooseBackup = transportRequest.isUsedRdma ? isBackup : false;
937 : HcclNetDevCtx netDevCtx;
938 0 : HcclResult ret = CreateDestSockets(
939 : tag, transportRequest.remoteUserRank, reqSingleSubComm.taskNum, connectSockets, netDevCtx,
940 0 : isInterRdma, transportRequest.isUsedRdma, chooseBackup, subCommIndex,
941 : transportRequest.linkType);
942 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[IncreAlloc]Create dest sockets failed"), ret);
943 :
944 0 : MachineType machineType = transportRequest.localUserRank < transportRequest.remoteUserRank ?
945 : MachineType::MACHINE_SERVER_TYPE :
946 : MachineType::MACHINE_CLIENT_TYPE;
947 : std::string threadStr
948 0 : = (isInterRdma ? "HcclTerL_" : "HcclIntra_") + std::to_string(threadsRapplyNum);
949 0 : bool isIndOp = false;
950 0 : IndOpMem indOpMem;
951 0 : linkThreads[threadsRapplyNum].reset(new (std::nothrow) std::thread(
952 0 : &TransportManager::CreateLink, this, tag, hrtErrMGetErrorContextPub(), machineType,
953 0 : rankInfoList_[userRank_].serverId, transportRequest.remoteUserRank,
954 0 : reqSingleSubComm.supportDataReceivedAck, reqSingleSubComm.linkMode,
955 0 : reqSingleSubComm.enableUseOneDoorbell, threadStr, connectSockets, inputMem, outputMem,
956 0 : transportRequest.isUsedRdma, std::ref(respSingleSubComm.links[rankIndex]), isAicpuModeEn,
957 0 : std::ref(linkResult[threadsRapplyNum]), netDevCtx, transportRequest.notifyNum, chooseBackup,
958 0 : isCapture, expMem, transportRequest.linkType, isIndOp, indOpMem, opType, false));
959 0 : CHK_SMART_PTR_NULL(linkThreads[threadsRapplyNum]); // 异常时其他线程待处理
960 0 : respSingleSubComm.status[rankIndex] = TransportStatus::READY; // 建链后 transport设置为ready状态
961 0 : threadsRapplyNum++;
962 0 : }
963 : }
964 0 : for (u32 index = 0; index < linkThreads.size(); index++) {
965 0 : if (linkThreads[index] != nullptr && linkThreads[index]->joinable()) {
966 0 : linkThreads[index]->join();
967 0 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
968 : }
969 : }
970 0 : linkThreads.clear();
971 0 : for (u32 index = 0; index < linkResult.size(); index++) {
972 0 : CHK_RET(linkResult[index]);
973 : }
974 0 : for (auto& tmpTag : socketTagVec_) {
975 0 : (void)socketManager_->DestroySockets(tmpTag);
976 : }
977 0 : socketTagVec_.clear();
978 0 : }
979 : }
980 0 : CHK_RET(notifyPool_->UnregisterOp(tag));
981 0 : return HCCL_SUCCESS;
982 0 : }
983 :
984 81 : bool TransportManager::IsHccsTransport(u32 remoteRank, TransportLinkType linkType)
985 : {
986 : // 判断p2p连接中，与remoteRank间的链路是否为hccs链路
987 81 : bool isHccs = true;
988 81 : if (linkType == TransportLinkType::RESERVED) {
989 : // 非hccs sio并发场景，直接通过获取底层优选链路类型来判断，获取失败时，默认为HCCS
990 77 : LinkTypeInServer linkTypeTmp = LinkTypeInServer::RESERVED_LINK_TYPE;
991 231 : HcclResult ret = hrtGetPairDeviceLinkType(
992 77 : rankInfoList_[userRank_].devicePhyId, rankInfoList_[remoteRank].devicePhyId, linkTypeTmp);
993 77 : if (ret != HCCL_SUCCESS) {
994 0 : HCCL_WARNING(
995 : "fail to get device link type for userRank[%u] remoteRank[%u] ret[%d], default to Hccs", userRank_,
996 : remoteRank, ret);
997 0 : return true;
998 : }
999 77 : if (linkTypeTmp == LinkTypeInServer::SIO_TYPE) {
1000 0 : isHccs = false;
1001 : }
1002 : } else { // 910_93 2 die concurrent
1003 : // hccs sio并发场景，直接通过linkType判断
1004 4 : isHccs = linkType == TransportLinkType::HCCS;
1005 : }
1006 :
1007 81 : return isHccs;
1008 : }
1009 :
1010 81 : HcclResult TransportManager::ConstructTransTag(
1011 : const std::string& tag, std::string& transTag, bool isInterRdma, u32 subCommIndex, bool isHccs)
1012 : {
1013 81 : transTag = (Is310PDevice() || isHaveCpuRank_) ? tag : identifier_ + "_res_optimize_" + std::to_string(subCommIndex);
1014 81 : if (isInterRdma) {
1015 0 : transTag += "_Inter_";
1016 : } else {
1017 81 : transTag += isHccs ? "_Hccs_" : "_SIO_";
1018 : }
1019 81 : return HCCL_SUCCESS;
1020 : }
1021 :
1022 76 : HcclResult TransportManager::GetIOMem(
1023 : const TransportIOMem& transMem, const TransportMemType inputMemType, const TransportMemType outputMemType,
1024 : DeviceMem& inputMem, DeviceMem& outputMem, DeviceMem& expMem)
1025 : {
1026 76 : if (inputMemType == CCL_INPUT) {
1027 47 : inputMem = transMem.cclInputMem;
1028 29 : } else if (inputMemType == SCRATCH) {
1029 12 : inputMem = transMem.scratchMem;
1030 17 : } else if (inputMemType == PARAM_INPUT) {
1031 18 : inputMem = transMem.paramInputMem;
1032 0 : } else if (inputMemType == AIV_INPUT) {
1033 0 : inputMem = transMem.aivInputMem;
1034 0 : } else if (inputMemType == AIV_OUTPUT) {
1035 0 : inputMem = transMem.aivOutputMem;
1036 0 : } else if (inputMemType == CCL_OUTPUT) {
1037 0 : inputMem = transMem.cclOutputMem;
1038 0 : } else if (inputMemType == USER_MEM) {
1039 0 : inputMem = transMem.userMem;
1040 : } else {
1041 0 : HCCL_ERROR("inputMemType is Invalid, inputMem not set");
1042 0 : return HCCL_E_INTERNAL;
1043 : }
1044 :
1045 77 : if (outputMemType == CCL_OUTPUT) {
1046 29 : outputMem = transMem.cclOutputMem;
1047 48 : } else if (outputMemType == SCRATCH) {
1048 18 : outputMem = transMem.scratchMem;
1049 30 : } else if (outputMemType == PARAM_OUTPUT) {
1050 12 : outputMem = transMem.paramOutputMem;
1051 18 : } else if (outputMemType == AIV_INPUT) {
1052 0 : outputMem = transMem.aivInputMem;
1053 18 : } else if (outputMemType == AIV_OUTPUT) {
1054 0 : outputMem = transMem.aivOutputMem;
1055 18 : } else if (outputMemType == CCL_INPUT) {
1056 6 : outputMem = transMem.cclInputMem;
1057 12 : } else if (outputMemType == PARAM_INPUT) {
1058 12 : outputMem = transMem.paramInputMem;
1059 0 : } else if (outputMemType == USER_MEM) {
1060 0 : outputMem = transMem.userMem;
1061 : } else {
1062 0 : HCCL_ERROR("outputMemType is Invalid, inputMem not set");
1063 0 : return HCCL_E_INTERNAL;
1064 : }
1065 :
1066 77 : expMem = transMem.expMem;
1067 76 : return HCCL_SUCCESS;
1068 : }
1069 :
1070 0 : u32 TransportManager::GetHostPort(s32 devicePhyId)
1071 : {
1072 0 : if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
1073 0 : return (devicePhyId + HOST_PARA_BASE_PORT);
1074 : } else {
1075 0 : return (devicePhyId + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM);
1076 : }
1077 : }
1078 :
1079 77 : u32 TransportManager::GetRemoteNicPort(s32 devicePhyId, u32 dstUserRank, bool isInterRdma)
1080 : {
1081 77 : DevType devType = DevType::DEV_TYPE_COUNT;
1082 77 : CHK_RET(hrtGetDeviceType(devType));
1083 154 : if ((nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST)
1084 77 : && !(
1085 0 : dstUserRank < nicRanksPort_.size() && nicRanksPort_[dstUserRank] != HCCL_INVALID_PORT
1086 0 : && devType == DevType::DEV_TYPE_910B)) {
1087 0 : return GetHostPort(devicePhyId);
1088 : }
1089 :
1090 : // isUseRankPort_在ranksPort初始化时一同配置：1. 异构场景 2. 开启device侧端口配置
1091 : // vnic port仅用于开启device侧端口配置时的sdma场景
1092 77 : bool useVnicPort = devPortSwitchOn_ && !isInterRdma && !Is310PDevice();
1093 77 : const std::vector<u32>& ranksPorts = useVnicPort ? vnicRanksPort_ : nicRanksPort_;
1094 77 : return GetNicPort(devicePhyId, ranksPorts, dstUserRank, isUseRankPort_);
1095 : }
1096 :
1097 81 : uint32_t TransportManager::GetConnectMode(RankId remoteRank)
1098 : {
1099 81 : NICDeployment localNicDeploy = NICDeployment::NIC_DEPLOYMENT_DEVICE;
1100 81 : NICDeployment remoteNicDeploy = NICDeployment::NIC_DEPLOYMENT_DEVICE;
1101 401 : for (auto it : rankInfoList_) {
1102 320 : if (it.userRank == remoteRank) {
1103 81 : remoteNicDeploy = it.nicDeploy;
1104 : }
1105 320 : if (it.userRank == userRank_) {
1106 81 : localNicDeploy = it.nicDeploy;
1107 : }
1108 320 : }
1109 81 : return static_cast<uint32_t>(localNicDeploy != remoteNicDeploy);
1110 : }
1111 :
1112 81 : HcclResult TransportManager::GetTransNewTag(
1113 : const std::string& tag, std::string& newTag, RankId remoteRank, bool& isInterRdma, u32 subCommIndex,
1114 : TransportLinkType linkType, HcclRankLinkInfo remoteLink, uint32_t mode)
1115 : {
1116 81 : if (mode != 0) {
1117 0 : if (userRank_ < remoteRank) {
1118 0 : newTag = identifier_ + "_" + std::to_string(userRank_) + "_" + std::to_string(remoteRank) + "_"
1119 0 : + devIpAddr_[0].GetReadableIP() + "_" + remoteLink.ip.GetReadableIP();
1120 : } else {
1121 0 : newTag = identifier_ + "_" + std::to_string(remoteRank) + "_" + std::to_string(userRank_) + "_"
1122 0 : + remoteLink.ip.GetReadableIP() + "_" + devIpAddr_[0].GetReadableIP();
1123 : }
1124 : } else {
1125 81 : bool isHccs = isInterRdma ? false : IsHccsTransport(remoteRank, linkType);
1126 81 : CHK_RET(ConstructTransTag(tag, newTag, isInterRdma, subCommIndex, isHccs));
1127 : }
1128 :
1129 81 : return HCCL_SUCCESS;
1130 : }
1131 :
1132 80 : HcclResult TransportManager::CreateDestSockets(
1133 : const std::string& tag, RankId remoteRank, u64 taskNum, std::vector<std::shared_ptr<HcclSocket>>& connectSockets,
1134 : HcclNetDevCtx& netDevCtx, bool& isInterRdma, bool forceRdma, bool isBackup, u32 subCommIndex,
1135 : TransportLinkType linkType)
1136 : {
1137 : // 改对端的ip和port
1138 80 : UpdateIsInterRdma(remoteRank, isInterRdma, forceRdma);
1139 81 : HCCL_INFO(
1140 : "[Create][DestSockets]UpdateIsInterRdma finished. local rank[%u], remote rank[%u],"
1141 : "isInterRdma[%d], forceRdma[%d], tag[%s]",
1142 : userRank_, remoteRank, isInterRdma, forceRdma, tag.c_str());
1143 :
1144 81 : u32 socketsPerLink = 1;
1145 81 : if (isInterRdma) {
1146 0 : if (!mulQpinfo_) {
1147 0 : mulQpinfo_.reset(static_cast<MulQpInfo*>(new (std::nothrow) MulQpInfo()));
1148 : }
1149 0 : CHK_PRT_RET(!mulQpinfo_, HCCL_ERROR("[Init][Transport]In create mulQpinfo failed"), HCCL_E_PTR);
1150 0 : CHK_RET(mulQpinfo_->Init(InitParams{
1151 : nicDeployment_, static_cast<std::int32_t>(rankInfoList_[userRank_].devicePhyId),
1152 : rankInfoList_[userRank_].deviceType}));
1153 0 : socketsPerLink = GetSocketsPerLink(taskNum, remoteRank);
1154 : }
1155 :
1156 81 : HcclRankLinkInfo remoteLinkInfo;
1157 80 : MakeRemoteLinkInfo(remoteRank, isInterRdma, socketsPerLink, remoteLinkInfo);
1158 81 : if (isBackup) {
1159 0 : remoteLinkInfo.ip = rankInfoList_[remoteRank].backupNicIp[0];
1160 0 : remoteLinkInfo.port = rankInfoList_[remoteRank].backupDevicePort == HCCL_INVALID_PORT ?
1161 : AICPU_RETRY_BACKUP_PORT :
1162 0 : rankInfoList_[remoteRank].backupDevicePort;
1163 : }
1164 :
1165 81 : HCCL_INFO(
1166 : "[%s] ip and port info. local rank[%u], remote rank[%u], isBackup[%d], port[%u], ip[%s]", __func__, userRank_,
1167 : remoteRank, isBackup, remoteLinkInfo.port, remoteLinkInfo.ip.GetReadableIP());
1168 :
1169 81 : std::string newTag;
1170 81 : uint32_t mode = GetConnectMode(remoteRank);
1171 81 : GetTransNewTag(tag, newTag, remoteRank, isInterRdma, subCommIndex, linkType, remoteLinkInfo, mode);
1172 :
1173 81 : HcclResult ret = HCCL_SUCCESS;
1174 81 : if (isInterRdma || Is310PDevice()) {
1175 0 : netDevCtx = ((nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE)
1176 0 : || (nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_HOST && !devIpAddr_[0].IsInvalid())) ?
1177 0 : netDevCtxMap_[devIpAddr_[0]] :
1178 0 : netDevCtxMap_[hostIp_];
1179 0 : if (isBackup && nicDeployment_ == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
1180 0 : netDevCtx = netDevCtxMap_[rankInfoList_[userRank_].backupNicIp[0]];
1181 0 : HCCL_DEBUG(
1182 : "[%s]refresh netDevCtx info. local rank[%u], remote rank[%u], isBackup[%d], port[%u], ip[%s]", __func__,
1183 : userRank_, remoteRank, isBackup, remoteLinkInfo.port,
1184 : (rankInfoList_[userRank_].backupNicIp[0]).GetReadableIP());
1185 : }
1186 0 : ret = socketManager_->CreateSingleLinkSocket(
1187 : newTag, netDevCtx, remoteLinkInfo, connectSockets, false, false, 0, mode);
1188 0 : if (!GetExternalInputHcclIsTcpMode()) {
1189 0 : std::vector<std::string>::iterator iter = std::find(socketTagVec_.begin(), socketTagVec_.end(), newTag);
1190 0 : if (iter == socketTagVec_.end()) {
1191 0 : socketTagVec_.push_back(newTag);
1192 : }
1193 : }
1194 : } else {
1195 : // server内非异构场景，使能P2P
1196 81 : bool isInterServer = false;
1197 81 : CHK_PRT(IsInterServer(remoteRank, isInterServer));
1198 :
1199 81 : if (!isInterServer && !isHaveCpuRank_) {
1200 77 : std::vector<u32> WaitP2PEnabledDevices;
1201 77 : WaitP2PEnabledDevices.push_back(rankInfoList_[remoteRank].devicePhyId);
1202 77 : HCCL_INFO(
1203 : "[Create][DestSockets]localDevicePhyId[%u] wait p2p enable with remoteDevicePhyId[%u]",
1204 : rankInfoList_[userRank_].devicePhyId, rankInfoList_[remoteRank].devicePhyId);
1205 77 : HcclResult ret = P2PMgmtPub::WaitP2PEnabled(WaitP2PEnabledDevices, [this]() -> bool {
1206 0 : return this->GetStopFlag();
1207 : });
1208 77 : if (ret != HCCL_SUCCESS) {
1209 0 : if (ret == HCCL_E_DRV) {
1210 0 : RankInfo loaclRankInfo = rankInfoList_[userRank_];
1211 0 : RankInfo remoteRankInfo = rankInfoList_[remoteRank];
1212 0 : DetectConnectionAnomalies::GetInstance(deviceLogicId_)
1213 0 : .AddIpQueue(loaclRankInfo, remoteRankInfo, NicType::VNIC_TYPE, deviceLogicId_);
1214 0 : }
1215 0 : CHK_PRT_RET(
1216 : true,
1217 : HCCL_ERROR(
1218 : "[Create][DestSockets]Wait Enable P2P Failed, src devicePhyId[%d], dst devicePhyId[%d], "
1219 : "ret[%u]",
1220 : rankInfoList_[userRank_].devicePhyId, rankInfoList_[remoteRank].devicePhyId, ret),
1221 : ret);
1222 : }
1223 77 : }
1224 81 : netDevCtx = netDevCtxMap_[localVnicIp_];
1225 81 : ret = socketManager_->CreateSingleLinkSocket(newTag, netDevCtx, remoteLinkInfo, connectSockets, false, true);
1226 : }
1227 81 : CHK_PRT_RET(
1228 : ret != HCCL_SUCCESS,
1229 : HCCL_ERROR(
1230 : "[Create][DestSockets]Create single link sockets failed, "
1231 : "local rank[%u], remote rank[%u], isInterRdma[%d]",
1232 : userRank_, remoteRank, isInterRdma),
1233 : ret);
1234 40 : return ret;
1235 81 : }
1236 :
1237 0 : u32 TransportManager::GetSocketsPerLink(u64 taskNum, u32 remoteRankId)
1238 : {
1239 0 : bool isEnableMulQp = false;
1240 0 : CHK_RET(mulQpinfo_->IsEnableMulQp(isEnableMulQp));
1241 0 : if (isEnableMulQp) {
1242 : PortNum portNum;
1243 0 : CHK_RET(mulQpinfo_->GetPortsNumByIpPair(
1244 : portNum, remoteRankId >= rankInfoList_.size() ?
1245 : KeyPair() :
1246 : std::make_pair(rankInfoList_[userRank_].nicIp[0], rankInfoList_[remoteRankId].nicIp[0])));
1247 0 : if (portNum > HCCL_QPS_PER_CONNECTION_DEFAULT) {
1248 0 : SetMultiQpMode(dispatcher_, true);
1249 0 : return MUL_QP_SOCKETS_PER_LINk;
1250 : }
1251 : }
1252 0 : u32 socketsPerLink = 1;
1253 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
1254 0 : if (taskNum == 0) {
1255 0 : taskNum = 1;
1256 : }
1257 0 : socketsPerLink = (taskNum + (HCCP_SQ_TEMPLATE_CAPACITY - 1)) / HCCP_SQ_TEMPLATE_CAPACITY;
1258 : }
1259 0 : return socketsPerLink;
1260 : }
1261 :
1262 0 : HcclResult TransportManager::CheckLinkNumAndSwitchLinkType(
1263 : TransportType& type, MachinePara& machinePara, const std::vector<std::shared_ptr<HcclSocket>> sockets)
1264 : {
1265 0 : u32 localCount = ibvCount_;
1266 0 : u32 remoteCount = 0;
1267 0 : CHK_RET(sockets[0]->Send(&localCount, sizeof(localCount)));
1268 0 : CHK_RET(sockets[0]->Recv(&remoteCount, sizeof(remoteCount)));
1269 0 : if (localCount > MASSIVE_IBV_CONNECTION_COUNT || remoteCount > MASSIVE_IBV_CONNECTION_COUNT) {
1270 : // 走aicpu直驱时暂时不支持iscapture特性
1271 0 : type = TransportType::TRANS_TYPE_DEVICE_DIRECT;
1272 : }
1273 0 : HCCL_INFO(
1274 : "[TransportManager][CheckLinkNumAndSwitchLinkType] local ibvCount[%u], remote IbvCount[%u] "
1275 : "localrank[%u], remoterank[%u], type[%d]",
1276 : localCount, remoteCount, machinePara.localUserrank, machinePara.remoteUserrank, type);
1277 0 : return HCCL_SUCCESS;
1278 : }
1279 :
1280 38 : HcclResult TransportManager::PrintErrorInfo(NicType nicType)
1281 : {
1282 : DevType devType;
1283 38 : CHK_RET(hrtGetDeviceType(devType));
1284 38 : if (devType != DevType::DEV_TYPE_910_93) {
1285 38 : return HCCL_SUCCESS;
1286 : }
1287 0 : std::string nicTypeStr;
1288 0 : switch (nicType) {
1289 0 : case NicType::VNIC_TYPE:
1290 0 : nicTypeStr = "VNIC_TYPE";
1291 0 : break;
1292 0 : case NicType::DEVICE_NIC_TYPE:
1293 0 : nicTypeStr = "DEVICE_NIC_TYPE";
1294 0 : break;
1295 0 : case NicType::HOST_NIC_TYPE:
1296 0 : nicTypeStr = "HOST_NIC_TYPE";
1297 0 : break;
1298 0 : default:
1299 0 : nicTypeStr = "unknown";
1300 : }
1301 : s64 phySuperPodId;
1302 0 : CHK_RET(hrtGetDeviceInfo(
1303 : deviceLogicId_, HcclRtDeviceModuleType::HCCL_RT_MODULE_TYPE_SYSTEM,
1304 : HcclRtDeviceInfoType::HCCL_INFO_TYPE_SUPER_POD_ID, phySuperPodId));
1305 0 : std::string logicSuperPodId = GetExternalInputLogicSuperPodId();
1306 0 : if (logicSuperPodId.empty()) {
1307 0 : HCCL_ERROR(
1308 : "[TransportManager][%s]local rank information: nicType[%s], logicSuperPodId is not set, "
1309 : "phySuperPodId[%lld].",
1310 : __func__, nicTypeStr.c_str(), phySuperPodId);
1311 : } else {
1312 0 : HCCL_ERROR(
1313 : "[TransportManager][%s]local rank information: nicType[%s], logicSuperPodId[%s], phySuperPodId[%lld]. "
1314 : "Note: Do not "
1315 : "configure ranks belonging to different physical superpod ID info a single logical superpod ID",
1316 : __func__, nicTypeStr.c_str(), logicSuperPodId.c_str(), phySuperPodId);
1317 : }
1318 0 : return HCCL_SUCCESS;
1319 0 : }
1320 :
1321 39 : HcclResult TransportManager::CreateLink(
1322 : const std::string& tag, const ErrContextPub& error_context, const MachineType machineType,
1323 : const std::string& serverId, const u32 remoteRank, const bool supportDataReceivedAck, const LinkMode linkMode,
1324 : const bool enableUseOneDoorbell, const std::string threadStr,
1325 : const std::vector<std::shared_ptr<HcclSocket>> sockets, const DeviceMem inputMem, const DeviceMem outputMem,
1326 : bool isUsedRdma, std::shared_ptr<Transport>& link, bool isAicpuModeEn, HcclResult& retOut,
1327 : const HcclNetDevCtx& netDevCtx, u32 notifyNum, bool isBackup, bool isCapture, const DeviceMem expMem,
1328 : TransportLinkType linkType, bool isIndOp, const IndOpMem indOpMem, const HcclCMDType& opType, bool isNpuDirectRoce)
1329 : {
1330 39 : hrtErrMSetErrorContextPub(error_context);
1331 : // 给当前线程添加名字
1332 40 : SetThreadName(threadStr);
1333 40 : link = nullptr;
1334 40 : retOut = hrtSetDevice(deviceLogicId_);
1335 40 : CHK_RET(retOut);
1336 :
1337 40 : SetWorkflowMode(workflowMode_); // 更新本线程的workflow
1338 :
1339 40 : MachinePara machinePara;
1340 40 : RankInfo loaclRankInfo = rankInfoList_[userRank_];
1341 40 : RankInfo remoteRankInfo = rankInfoList_[remoteRank];
1342 40 : HcclResult ret = HCCL_SUCCESS;
1343 : do {
1344 40 : ret = SetMachinePara(
1345 : tag, machineType, serverId, remoteRank, supportDataReceivedAck, linkMode, sockets, inputMem, outputMem,
1346 : expMem, isAicpuModeEn, isBackup, isCapture, notifyNum, trafficClass_, serviceLevel_, machinePara,
1347 : loaclRankInfo, remoteRankInfo, netDevCtx, linkType, indOpMem, isIndOp, opType, isNpuDirectRoce);
1348 40 : retOut = ret;
1349 40 : std::string tmpErrInfo = ret == HCCL_E_TIMEOUT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
1350 40 : CHK_PRT_BREAK(
1351 : ret != HCCL_SUCCESS,
1352 : HCCL_ERROR(
1353 : "[%s][%s][%s]SetMachinePara error.", __func__, LOG_KEYWORDS_INIT_CHANNEL.c_str(),
1354 : tmpErrInfo.c_str()), );
1355 :
1356 2 : HCCL_INFO(
1357 : "inputMem[%p],outputMem[%p], inputMem size[%llu], outputMem size[%llu]", inputMem.ptr(), outputMem.ptr(),
1358 : inputMem.size(), outputMem.size());
1359 2 : if (isIndOp) {
1360 0 : HCCL_DEBUG(
1361 : "userHostMem num[%llu], userDeviceMem num[%llu]", indOpMem.userHostMem.size(),
1362 : indOpMem.userDeviceMem.size());
1363 : }
1364 2 : HCCL_INFO(
1365 : "[createLink para]tag[%s], rank[%u]-localUserrank[%u]-localIpAddr[%s], linkMode[%d] "
1366 : "dst_rank[%u]-remoteUserrank[%u]-remote_ip_addr[%s], machineType[%d], serverId[%s], "
1367 : "nicDeploy[%d], isBackup[%d], opType[%d]",
1368 : tag.c_str(), userRank_, rankInfoList_[userRank_].worldRank, rankInfoList_[userRank_].serverId.c_str(),
1369 : machinePara.linkMode, remoteRank, rankInfoList_[remoteRank].worldRank,
1370 : rankInfoList_[remoteRank].serverId.c_str(), machinePara.machineType, machinePara.serverId.c_str(),
1371 : machinePara.nicDeploy, isBackup, opType);
1372 : // transport初始化
1373 2 : TransportType type = TransportType::TRANS_TYPE_RESERVED;
1374 2 : CHK_PRT(GetTransportType(remoteRank, isUsedRdma, type));
1375 : // A2/A3 batch_send_recv 走roce才切换到新链路
1376 2 : if (type == TransportType::TRANS_TYPE_IBV_EXP && opType_ == HCCL_CMD_BATCH_SEND_RECV) {
1377 0 : ret = CheckLinkNumAndSwitchLinkType(type, machinePara, sockets);
1378 0 : retOut = ret;
1379 0 : CHK_PRT_BREAK(
1380 : ret != HCCL_SUCCESS,
1381 : HCCL_ERROR(
1382 : "[%s]errNo[0x%016llx]CheckLinkNumAndSwitchLinkType error.", __func__, HCCL_ERROR_CODE(ret)), );
1383 0 : std::lock_guard<std::mutex> lock(ibvCountMutex_);
1384 : // 之前已经经过链了，则使用之前的老链路，bsr会建2条链路
1385 0 : if (remoteTransportMap_.find(machinePara.remoteUserrank) != remoteTransportMap_.end()) {
1386 0 : type = remoteTransportMap_[machinePara.remoteUserrank];
1387 0 : HCCL_INFO(
1388 : "[TransportManager][CreateLink] use the same type as before, localRank %u remoteRank %u type %d",
1389 : machinePara.localUserrank, machinePara.remoteUserrank, type);
1390 : } else {
1391 0 : remoteTransportMap_.insert(std::make_pair(machinePara.remoteUserrank, type));
1392 0 : HCCL_INFO(
1393 : "[TransportManager][CreateLink] transportMap save remoterank %u type %d",
1394 : machinePara.remoteUserrank, type);
1395 : }
1396 0 : if (type == TransportType::TRANS_TYPE_IBV_EXP) {
1397 0 : ibvCount_++;
1398 0 : } else if (type == TransportType::TRANS_TYPE_DEVICE_DIRECT) {
1399 0 : machinePara.qpMode = QPMode::NORMAL;
1400 0 : machinePara.queueDepthAttr.recvCqDepth = RECV_QP_DEPTH_FOR_BSR;
1401 0 : machinePara.queueDepthAttr.rqDepth = RECV_QP_DEPTH_FOR_BSR;
1402 0 : machinePara.queueDepthAttr.sendCqDepth = SEND_QP_DEPTH_FOR_BSR;
1403 0 : machinePara.queueDepthAttr.sqDepth = SEND_QP_DEPTH_FOR_BSR;
1404 : }
1405 0 : }
1406 2 : ret = TransportInit(machinePara, link, enableUseOneDoorbell, type);
1407 2 : retOut = ret;
1408 2 : CHK_PRT_BREAK(
1409 : ret != HCCL_SUCCESS,
1410 : HCCL_ERROR("[%s]errNo[0x%016llx]TransportInit error.", __func__, HCCL_ERROR_CODE(ret)), );
1411 40 : } while (0);
1412 :
1413 40 : if (ret != HCCL_SUCCESS) {
1414 38 : link = nullptr;
1415 38 : retOut = ret;
1416 38 : if (ret == HCCL_E_MEMORY) {
1417 : std::string err_str
1418 : = "[Create][DestLink]Transport init error! IPC memory allocation failed due to "
1419 : "possible memory limit exceeded. Suggested solution: Use 3TB / (ranksize * 2) as the upper limit of "
1420 0 : "HCCL_BUFFSIZE.";
1421 0 : HCCL_ERROR(
1422 : "[%s][%s]%s", LOG_KEYWORDS_INIT_CHANNEL.c_str(), LOG_KEYWORDS_RUN_FAILED.c_str(), err_str.c_str());
1423 0 : }
1424 :
1425 38 : NicType nicType = sockets[0]->GetSocketType();
1426 38 : DetectConnectionAnomalies::GetInstance(deviceLogicId_)
1427 38 : .AddIpQueue(loaclRankInfo, remoteRankInfo, nicType, deviceLogicId_);
1428 :
1429 : char errorLogBuffer[LOG_TMPBUF_SIZE];
1430 114 : s32 stringRet = snprintf_s(
1431 : errorLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
1432 : "createLink para:rank[%u]-localUserrank[%u]-localIpAddr[%s/%d], "
1433 : "remoteRank[%u]-remoteUserrank[%u]-remoteIpAddr[%s/%d], "
1434 : "machineType[%d], linkMode[%d], isUsedRdma[%d], tag[%s]",
1435 38 : userRank_, rankInfoList_[userRank_].worldRank, rankInfoList_[userRank_].serverId.c_str(),
1436 38 : rankInfoList_[userRank_].devicePhyId, remoteRank, rankInfoList_[remoteRank].worldRank,
1437 38 : rankInfoList_[remoteRank].serverId.c_str(), rankInfoList_[remoteRank].devicePhyId, machinePara.machineType,
1438 : machinePara.linkMode, isUsedRdma, machinePara.tag.c_str());
1439 38 : CHK_PRT_CONT(stringRet == -1, HCCL_ERROR("[Create][DestLink]Transport init error! Failed to build log info"));
1440 38 : std::string tmpErrInfo = ret == HCCL_E_TIMEOUT ? LOG_KEYWORDS_TIMEOUT : LOG_KEYWORDS_RUN_FAILED;
1441 38 : HCCL_ERROR(
1442 : "[%s][%s]Transport init error! %s", LOG_KEYWORDS_INIT_CHANNEL.c_str(), tmpErrInfo.c_str(), errorLogBuffer);
1443 38 : CHK_PRT(PrintErrorInfo(nicType));
1444 38 : return ret;
1445 38 : }
1446 2 : HCCL_INFO(
1447 : "[createLink success]:rank[%u]-localUserrank[%u]-localIpAddr[%s], "
1448 : "dst_rank[%u]-remoteUserrank[%u]-remote_ip_addr[%s], tag[%s]",
1449 : userRank_, rankInfoList_[userRank_].worldRank, rankInfoList_[userRank_].serverId.c_str(), remoteRank,
1450 : rankInfoList_[remoteRank].worldRank, rankInfoList_[remoteRank].serverId.c_str(), machinePara.tag.c_str());
1451 :
1452 2 : return HCCL_SUCCESS;
1453 40 : }
1454 :
1455 42 : HcclResult TransportManager::SetMachinePara(
1456 : const std::string& tag, MachineType machineType, const std::string& serverId, u32 dstRank,
1457 : const bool supportDataReceivedAck, const LinkMode linkMode,
1458 : const std::vector<std::shared_ptr<HcclSocket>>& socketList, const DeviceMem& inputMem, const DeviceMem& outputMem,
1459 : const DeviceMem& expMem, bool isAicpuModeEn, bool isBackup, bool isCapture, u32 notifyNum, u32 trafficClass,
1460 : u32 serviceLevel, MachinePara& machinePara, RankInfo& loaclRank, RankInfo& remoteRank,
1461 : const HcclNetDevCtx& netDevCtx, TransportLinkType linkType, const IndOpMem& indOpMem, bool isIndOp,
1462 : const HcclCMDType& opType, bool isNpuDirectRoce)
1463 : {
1464 42 : machinePara.notifyNum = notifyNum;
1465 42 : machinePara.linkMode = linkMode;
1466 42 : machinePara.machineType = machineType;
1467 42 : machinePara.serverId = serverId;
1468 42 : machinePara.localUserrank = rankInfoList_[userRank_].userRank;
1469 42 : machinePara.remoteUserrank = rankInfoList_[dstRank].userRank;
1470 42 : machinePara.localWorldRank = rankInfoList_[userRank_].worldRank;
1471 42 : machinePara.remoteWorldRank = rankInfoList_[dstRank].worldRank;
1472 42 : machinePara.collectiveId = identifier_;
1473 42 : machinePara.deviceType = static_cast<DevType>(rankInfoList_[dstRank].deviceType);
1474 42 : machinePara.inputMem = inputMem;
1475 42 : machinePara.outputMem = outputMem;
1476 42 : machinePara.tc = trafficClass;
1477 42 : machinePara.sl = serviceLevel;
1478 42 : if (expMem.ptr() != nullptr) {
1479 29 : machinePara.mem.push_back(expMem);
1480 : } else {
1481 13 : machinePara.mem.clear();
1482 : }
1483 42 : if (isIndOp) {
1484 0 : machinePara.userHostMem = indOpMem.userHostMem;
1485 0 : machinePara.userDeviceMem = indOpMem.userDeviceMem;
1486 : }
1487 42 : machinePara.isIndOp = isIndOp;
1488 42 : machinePara.linkAttribute = 0x03; /* 0x03同时支持目的端和源端发起 */
1489 42 : machinePara.tag = tag;
1490 42 : if (isBackup) {
1491 0 : machinePara.localIpAddr = rankInfoList_[userRank_].backupNicIp[0];
1492 0 : machinePara.remoteIpAddr = rankInfoList_[dstRank].backupNicIp[0];
1493 : u32 localDevBackUpPhyId;
1494 0 : CHK_RET(hrtGetPairDevicePhyId(rankInfoList_[userRank_].devicePhyId, localDevBackUpPhyId));
1495 0 : machinePara.localDeviceId = static_cast<s32>(localDevBackUpPhyId);
1496 : u32 remoteDevBackUpPhyId;
1497 0 : CHK_RET(hrtGetPairDevicePhyId(rankInfoList_[dstRank].devicePhyId, remoteDevBackUpPhyId));
1498 0 : machinePara.remoteDeviceId = static_cast<s32>(remoteDevBackUpPhyId);
1499 0 : HCCL_DEBUG(
1500 : "[%s]isBackup[%d], machinePara.localIpAddr[%s], machinePara.remoteIpAddr[%s], "
1501 : "machinePara.localDeviceId[%d], machinePara.remoteDeviceId[%d].",
1502 : __func__, isBackup, machinePara.localIpAddr.GetReadableIP(), machinePara.remoteIpAddr.GetReadableIP(),
1503 : machinePara.localDeviceId, machinePara.remoteDeviceId);
1504 : } else {
1505 42 : machinePara.localIpAddr = rankInfoList_[userRank_].nicIp[0];
1506 42 : machinePara.remoteIpAddr = rankInfoList_[dstRank].nicIp[0];
1507 41 : machinePara.localDeviceId = rankInfoList_[userRank_].devicePhyId;
1508 41 : machinePara.remoteDeviceId = rankInfoList_[dstRank].devicePhyId;
1509 41 : HCCL_DEBUG(
1510 : "[%s]isBackup[%d], machinePara.localIpAddr[%s], machinePara.remoteIpAddr[%s], "
1511 : "machinePara.localDeviceId[%d], machinePara.remoteDeviceId[%d].",
1512 : __func__, isBackup, machinePara.localIpAddr.GetReadableIP(), machinePara.remoteIpAddr.GetReadableIP(),
1513 : machinePara.localDeviceId, machinePara.remoteDeviceId);
1514 : }
1515 : // 把原来的两层vector变成一层, 方便后继调用
1516 42 : if (socketList.size() > 0) {
1517 38 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> socketsMap;
1518 38 : socketsMap[dstRank] = socketList;
1519 38 : std::map<u32, u32> dstRankToUserRank;
1520 38 : dstRankToUserRank[dstRank] = dstRank;
1521 38 : CHK_RET(socketManager_->WaitLinksEstablishCompleted(
1522 : socketList[0]->GetLocalRole(), socketsMap, dstRankToUserRank, loaclRank, remoteRank, netDevCtx));
1523 0 : machinePara.sockets = socketList;
1524 76 : }
1525 4 : machinePara.exchangeInfo.resize(rankConsistentDataLength_);
1526 4 : CHK_RET(RankConsistentcyChecker::GetInstance().GetCheckFrame(
1527 : &machinePara.exchangeInfo[0], rankConsistentDataLength_, tag));
1528 4 : machinePara.supportDataReceivedAck = supportDataReceivedAck; /* NeedDataReceivedAck(); */
1529 4 : machinePara.nicDeploy = nicDeployment_;
1530 4 : machinePara.localSocketPort = rankInfoList_[userRank_].hostPort;
1531 4 : machinePara.remoteSocketPort = rankInfoList_[dstRank].hostPort;
1532 4 : if (isBackup) {
1533 : u32 tempDevBackUpPhyId;
1534 0 : CHK_RET(hrtGetPairDevicePhyId(rankInfoList_[userRank_].devicePhyId, tempDevBackUpPhyId));
1535 : u32 tempDevBackUpLogicId;
1536 0 : CHK_RET(hrtGetDeviceIndexByPhyId(tempDevBackUpPhyId, tempDevBackUpLogicId));
1537 0 : machinePara.deviceLogicId = static_cast<s32>(tempDevBackUpLogicId);
1538 0 : HCCL_DEBUG("[%s]isBackup[%d], machinePara.deviceLogicId[%d].", __func__, isBackup, machinePara.deviceLogicId);
1539 : } else {
1540 4 : machinePara.deviceLogicId = deviceLogicId_;
1541 4 : HCCL_DEBUG("[%s]isBackup[%d], machinePara.deviceLogicId[%d].", __func__, isBackup, machinePara.deviceLogicId);
1542 : }
1543 :
1544 4 : machinePara.srcPorts = std::vector<std::uint16_t>(1, 0); /* 默认填充一个元素，0代表默认不配置 */
1545 4 : machinePara.isAicpuModeEn = isAicpuModeEn;
1546 4 : if (linkType == TransportLinkType::RESERVED) {
1547 : // 非910_93 2die sio与hccs并发场景，specifyLink设置为RESERVED_LINK_TYPE，平台层将按实际链路类型建链
1548 2 : machinePara.specifyLink = LinkTypeInServer::RESERVED_LINK_TYPE;
1549 : } else {
1550 : // 910_93 2die sio与hccs并发场景，
1551 : // 并发链路中的的hccs链路specifyLink设置为HCCS_SW_TYPE，平台层将使用hccs链路来建链；
1552 : // 并发链路中的的sio链路specifyLink设置为SIO_TYPE，平台层将使用sio链路来建链
1553 : machinePara.specifyLink
1554 2 : = (linkType == TransportLinkType::SIO) ? LinkTypeInServer::SIO_TYPE : LinkTypeInServer::HCCS_SW_TYPE;
1555 : }
1556 :
1557 4 : if (isCapture) {
1558 0 : machinePara.qpMode = QPMode::OFFLOAD;
1559 : }
1560 :
1561 4 : if (isNpuDirectRoce) {
1562 : // AIV ROCE直驱场景，需要将QPMode更改为NORMAL模式，以避免底层走入stars调度的下发流程
1563 1 : machinePara.qpMode = QPMode::NORMAL;
1564 : }
1565 :
1566 : // reduce相关算子需要使能atomic write能力，用于实现rdma wqe(reduce+record)保序
1567 4 : bool isReduceOp = (opType == HCCL_CMD_ALLREDUCE) || (opType == HCCL_CMD_REDUCE)
1568 8 : || (opType == HCCL_CMD_REDUCE_SCATTER) || (opType == HCCL_CMD_REDUCE_SCATTER_V);
1569 4 : bool isSupportAtomicWrite = false;
1570 4 : CHK_RET(IsSupportAtomicWrite(machinePara.deviceType, machinePara.localDeviceId, isSupportAtomicWrite));
1571 4 : machinePara.enableAtomicWrite = isSupportAtomicWrite && isReduceOp;
1572 4 : HCCL_DEBUG(
1573 : "%s enableAtomicWrite[%d], opType[%d], isSupportAtomicWrite[%d]", __func__, machinePara.enableAtomicWrite,
1574 : opType, isSupportAtomicWrite);
1575 4 : return HCCL_SUCCESS;
1576 : }
1577 :
1578 2 : HcclResult TransportManager::GetTransportType(const u32 dstRank, bool isUsedRdma, TransportType& transportType)
1579 : {
1580 : // 判断是否在同一个server
1581 2 : bool isInterServer = false;
1582 2 : CHK_PRT(IsInterServer(dstRank, isInterServer));
1583 :
1584 2 : if (!isInterServer) {
1585 0 : LinkTypeInServer linkType = LinkTypeInServer::RESERVED_LINK_TYPE;
1586 0 : CHK_RET(hrtGetPairDeviceLinkType(
1587 : rankInfoList_[userRank_].devicePhyId, rankInfoList_[dstRank].devicePhyId, linkType));
1588 0 : if (isUsedRdma) {
1589 0 : transportType = TransportType::TRANS_TYPE_IBV_EXP;
1590 : } else {
1591 0 : transportType = TransportType::TRANS_TYPE_P2P;
1592 : }
1593 : } else { // server间
1594 2 : if ((!isUsedRdma) && IsSupportInterHccs(dstRank)) {
1595 : // 超节点内节点间走HCCS通信
1596 0 : transportType = TransportType::TRANS_TYPE_P2P;
1597 : } else {
1598 2 : transportType = TransportType::TRANS_TYPE_IBV_EXP;
1599 : }
1600 : }
1601 :
1602 2 : HCCL_INFO("GetTransportType: srcRank[%u], dstRank[%u], transport_type[%d].", userRank_, dstRank, transportType);
1603 2 : return HCCL_SUCCESS;
1604 : }
1605 :
1606 0 : void TransportManager::SetTransportParam(TransportPara& para)
1607 : {
1608 0 : std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
1609 0 : para.timeout = kdefaultTimeout;
1610 0 : para.virtualFlag = false;
1611 0 : }
1612 :
1613 0 : HcclResult TransportManager::TransportInit(
1614 : MachinePara& machinePara, std::shared_ptr<Transport>& link, bool useOneDoorbell, TransportType type)
1615 : {
1616 : // 实例化TransportBase
1617 0 : TransportPara para{};
1618 0 : SetTransportParam(para);
1619 :
1620 0 : if (type == TransportType::TRANS_TYPE_P2P) {
1621 0 : link.reset(new (std::nothrow) Transport(type, para, dispatcher_, notifyPool_, machinePara));
1622 0 : } else if (type == TransportType::TRANS_TYPE_IBV_EXP) {
1623 0 : bool isEnableMulQp = false;
1624 0 : CHK_RET(mulQpinfo_->IsEnableMulQp(isEnableMulQp));
1625 0 : if (isEnableMulQp) {
1626 0 : CHK_RET(mulQpinfo_->GetSpecialSourcePortsByIpPair(
1627 : machinePara.srcPorts, std::make_pair(machinePara.localIpAddr, machinePara.remoteIpAddr)));
1628 : }
1629 0 : link.reset(new (std::nothrow) Transport(type, para, dispatcher_, notifyPool_, machinePara));
1630 0 : } else if (type == TransportType::TRANS_TYPE_DEVICE_DIRECT) {
1631 0 : bool isEnableMulQp = false;
1632 0 : CHK_RET(mulQpinfo_->IsEnableMulQp(isEnableMulQp));
1633 0 : if (isEnableMulQp) {
1634 0 : CHK_RET(mulQpinfo_->GetSpecialSourcePortsByIpPair(
1635 : machinePara.srcPorts, std::make_pair(machinePara.localIpAddr, machinePara.remoteIpAddr)));
1636 : }
1637 0 : link.reset(new (std::nothrow) Transport(type, para, dispatcher_, notifyPool_, machinePara));
1638 : } else {
1639 0 : HCCL_ERROR("[Init][Transport]not supported transport type");
1640 0 : return HCCL_E_NOT_SUPPORT;
1641 : }
1642 :
1643 0 : CHK_PRT_RET(!link, HCCL_ERROR("[Init][Transport]In create link, new link failed"), HCCL_E_PTR);
1644 :
1645 0 : if (useOneDoorbell) {
1646 0 : link->EnableUseOneDoorbell();
1647 : }
1648 :
1649 0 : CHK_RET(link->Init());
1650 : // 算子一致性校验
1651 0 : std::vector<u8> recvData = link->GetExchangeInfo();
1652 0 : if (recvData.size() != 0) {
1653 0 : CHK_PRT_RET(
1654 : recvData.size() != machinePara.exchangeInfo.size(),
1655 : HCCL_ERROR(
1656 : "[Check][ExchangeInfo]remote exchangInfo size[%zu], local exchangeInfo size[%zu]", recvData.size(),
1657 : machinePara.exchangeInfo.size()),
1658 : HCCL_E_INTERNAL);
1659 0 : CHK_RET(RankConsistentcyChecker::GetInstance().CheckFrameRecv(
1660 : &recvData[0], recvData.size(), machinePara.tag.c_str()));
1661 : }
1662 0 : return HCCL_SUCCESS;
1663 0 : }
1664 :
1665 151 : bool TransportManager::IsSupportInterHccs(const u32 dstRank)
1666 : {
1667 : // 仅判断超节点内, 兼容打平通信域同时有server内和server间, 因此不判断server_id
1668 151 : bool isInterHccsDisable = GetExternalInputInterHccsDisable();
1669 151 : const std::string& curSuperPodId = rankInfoList_[userRank_].superPodId;
1670 151 : const std::string& dstSuperPodId = rankInfoList_[dstRank].superPodId;
1671 :
1672 151 : bool isInterHccs = isInterHccsDisable == false && useSuperPodMode_ == true && curSuperPodId.empty() == false
1673 302 : && curSuperPodId == dstSuperPodId;
1674 :
1675 151 : HCCL_INFO(
1676 : "[IsSupportInterHccs] rank[%u], superPodId[%s], dstRank[%u], dstSuperPodId[%s], useSuperPodMode[%d], "
1677 : "isInterHccsDisable[%d], isInterHccs[%d]",
1678 : userRank_, curSuperPodId.c_str(), dstRank, dstSuperPodId.c_str(), useSuperPodMode_, isInterHccsDisable,
1679 : isInterHccs);
1680 156 : return isInterHccs;
1681 : }
1682 :
1683 149 : void TransportManager::UpdateIsInterRdma(const u32 remoteRank, bool& isInterRdma, bool forceRdma) // 待确认判断是否完善
1684 : {
1685 : // 超节点内节点间采用HCCS通信的, 放至dstIntraClientVec_, 采用p2p建链
1686 149 : bool isInterHccs = IsSupportInterHccs(remoteRank);
1687 154 : if (isInterHccs && (!forceRdma)) {
1688 0 : isInterRdma = false;
1689 154 : } else if (rankInfoList_[userRank_].serverId != rankInfoList_[remoteRank].serverId) {
1690 0 : isInterRdma = true;
1691 : } else {
1692 : LinkTypeInServer linkType;
1693 154 : hrtGetPairDeviceLinkType(rankInfoList_[userRank_].devicePhyId, rankInfoList_[remoteRank].devicePhyId, linkType);
1694 154 : isInterRdma = (isUsedRdmaLevel0_ && linkType == LinkTypeInServer::PXI_TYPE) || forceRdma;
1695 : }
1696 154 : }
1697 :
1698 77 : HcclResult TransportManager::MakeRemoteLinkInfo(
1699 : const u32 remoteRank, bool isInterRdma, u32 socketsPerLink, HcclRankLinkInfo& remoteLinkInfo)
1700 : {
1701 77 : RankInfo dstRankInfo = rankInfoList_[remoteRank];
1702 77 : remoteLinkInfo.userRank = dstRankInfo.userRank;
1703 77 : remoteLinkInfo.devicePhyId = dstRankInfo.devicePhyId;
1704 77 : if (isInterRdma || Is310PDevice()) {
1705 0 : remoteLinkInfo.ip = dstRankInfo.nicIp[0];
1706 0 : remoteLinkInfo.port = GetRemoteNicPort(remoteLinkInfo.devicePhyId, dstRankInfo.userRank, isInterRdma);
1707 0 : remoteLinkInfo.socketsPerLink = socketsPerLink;
1708 : } else {
1709 77 : remoteLinkInfo.ip = HcclIpAddress(dstRankInfo.devicePhyId);
1710 77 : if (useSuperPodMode_) {
1711 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
1712 : rankInfoList_[userRank_].devicePhyId, DeviceIdType::DEVICE_ID_TYPE_SDID,
1713 : rankInfoList_[remoteRank].superDeviceId, remoteLinkInfo.ip));
1714 : } else {
1715 77 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
1716 : rankInfoList_[userRank_].devicePhyId, DeviceIdType::DEVICE_ID_TYPE_PHY_ID,
1717 : rankInfoList_[remoteRank].devicePhyId, remoteLinkInfo.ip));
1718 : }
1719 231 : remoteLinkInfo.port = GetRemoteNicPort(
1720 77 : rankInfoList_[remoteRank].devicePhyId, rankInfoList_[remoteRank].userRank, isInterRdma); // ?
1721 77 : remoteLinkInfo.socketsPerLink = socketsPerLink;
1722 : }
1723 77 : HCCL_INFO(
1724 : "[TransportManager][MakeRemoteLinkInfo] isInterRdma[%u], is310PDevice[%u], "
1725 : "remote rank: userRank[%u], devPhyId[%u], ip[%s], port[%u], socketsPerLink[%u]",
1726 : isInterRdma, Is310PDevice(), remoteLinkInfo.userRank, remoteLinkInfo.devicePhyId,
1727 : remoteLinkInfo.ip.GetReadableAddress(), remoteLinkInfo.port, remoteLinkInfo.socketsPerLink);
1728 77 : return HCCL_SUCCESS;
1729 77 : }
1730 :
1731 470 : HcclResult TransportManager::SetStopFlag(bool value)
1732 : {
1733 470 : stopFlag_.store(value);
1734 470 : return HCCL_SUCCESS;
1735 : }
1736 :
1737 1047 : void TransportManager::SetIsStandardCard(bool isStandardCard)
1738 : {
1739 1047 : isStandardCard_ = isStandardCard;
1740 1047 : return;
1741 : }
1742 :
1743 17 : bool TransportManager::GetStopFlag() { return stopFlag_.load(); }
1744 :
1745 1048 : void TransportManager::SetPortConfig(bool devPortSwitchOn) { devPortSwitchOn_ = devPortSwitchOn; }
1746 :
1747 111 : void TransportManager::SetOpType(HcclCMDType opType)
1748 : {
1749 111 : opType_ = opType;
1750 111 : return;
1751 : }
1752 :
1753 0 : HcclResult TransportManager::SetGroupMode(bool groupMode)
1754 : {
1755 0 : isGroupMode_ = groupMode;
1756 0 : HCCL_INFO("[SetGroupMode] isGroupMode_=[%d]", isGroupMode_);
1757 0 : return HCCL_SUCCESS;
1758 : }
1759 :
1760 59 : std::map<u32, TransportType> TransportManager::GetRemoteTransportMap() { return remoteTransportMap_; }
1761 :
1762 83 : HcclResult TransportManager::IsInterServer([[maybe_unused]] const u32 dstRank, [[maybe_unused]] bool& isInterServer)
1763 : {
1764 : #if !defined(CCL_KERNEL_AICPU) && !defined(HCCD)
1765 83 : if (rankInfoList_[userRank_].deviceType == DevType::DEV_TYPE_910_93) {
1766 0 : uint32_t userRankServerId = 0;
1767 0 : uint32_t remoteRankServerId = 0;
1768 0 : rtError_t ret = rtGetServerIDBySDID(rankInfoList_[userRank_].superDeviceId, &userRankServerId);
1769 0 : CHK_PRT_RET(
1770 : ret != RT_ERROR_NONE,
1771 : HCCL_ERROR(
1772 : "[IsInterServer]rtGetServerIDBySDID failed sdid[0x%08x], serverID[%u], ret[%u]",
1773 : rankInfoList_[userRank_].superDeviceId, userRankServerId, ret),
1774 : HCCL_E_RUNTIME);
1775 :
1776 0 : ret = rtGetServerIDBySDID(rankInfoList_[dstRank].superDeviceId, &remoteRankServerId);
1777 0 : CHK_PRT_RET(
1778 : ret != RT_ERROR_NONE,
1779 : HCCL_ERROR(
1780 : "[IsInterServer]rtGetServerIDBySDID failed sdid[0x%08x], serverID[%u], ret[%u]",
1781 : rankInfoList_[dstRank].superDeviceId, remoteRankServerId, ret),
1782 : HCCL_E_RUNTIME);
1783 0 : isInterServer = (userRankServerId != remoteRankServerId)
1784 0 : || (rankInfoList_[userRank_].superPodId != rankInfoList_[dstRank].superPodId);
1785 0 : HCCL_INFO(
1786 : "[IsInterServer]localSDID[0x%08x], localdevicePhyId[%d], localServerId[%s], localServerIdBySDID[%d], "
1787 : "localSuperPodId[%s], "
1788 : "remoteSDID[0x%08x], remotedevicePhyId[%d], remoteRankServerId[%s], remoteServerIdBySDID[%d], "
1789 : "remoteSuperPodId[%s], "
1790 : "isInterServer[%s]",
1791 : rankInfoList_[userRank_].superDeviceId, rankInfoList_[userRank_].devicePhyId,
1792 : rankInfoList_[userRank_].serverId.c_str(), userRankServerId, rankInfoList_[userRank_].superPodId.c_str(),
1793 : rankInfoList_[dstRank].superDeviceId, rankInfoList_[dstRank].devicePhyId,
1794 : rankInfoList_[dstRank].serverId.c_str(), remoteRankServerId, rankInfoList_[dstRank].superPodId.c_str(),
1795 : isInterServer ? "true" : "false");
1796 : } else {
1797 83 : isInterServer = rankInfoList_[userRank_].serverId != rankInfoList_[dstRank].serverId;
1798 83 : HCCL_INFO(
1799 : "[IsInterServer]localdevicePhyId[%d], localServerId[%s], "
1800 : "remotedevicePhyId[%d], remoteServerId[%s], isInterServer[%s]",
1801 : rankInfoList_[userRank_].devicePhyId, rankInfoList_[userRank_].serverId.c_str(),
1802 : rankInfoList_[dstRank].devicePhyId, rankInfoList_[dstRank].serverId.c_str(),
1803 : isInterServer ? "true" : "false");
1804 : }
1805 83 : return HCCL_SUCCESS;
1806 : #else
1807 : HCCL_ERROR("[IsInterServer]Does not support this interface.");
1808 : return HCCL_E_NOT_SUPPORT;
1809 : #endif
1810 : }
1811 : } // namespace hccl
|
