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