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 "comm_base.h"
12 : #include <arpa/inet.h>
13 : #include <securec.h>
14 :
15 : #include "externalinput_pub.h"
16 : #include "hccl_common.h"
17 : #include "device_capacity.h"
18 : #include "p2p_mgmt_pub.h"
19 : #include "rank_consistentcy_checker.h"
20 :
21 : namespace hccl {
22 : constexpr s32 HCCL_DEFAULT_INITIAL_VALUE = -1;
23 28 : CommBase::CommBase(
24 : const std::string& collectiveId, const u32 userRank, const u32 userRankSize, const u32 rank, const u32 rankSize,
25 : const std::vector<RankInfo> paraVector, const TopoType topoFlag, const HcclDispatcher dispatcher,
26 : const std::unique_ptr<NotifyPool>& notifyPool, std::map<HcclIpAddress, HcclNetDevCtx>& netDevCtxMap,
27 : const IntraExchanger& exchanger, const DeviceMem& inputMem, const DeviceMem& outputMem, const bool isUsedRdmaLevel0,
28 : const std::string& tag, const NICDeployment nicDeployInner, bool isAlltoAllCommMesh, const bool useOneDoorbell,
29 : const bool isAicpuModeEn, const u32 rankRoot, const bool isHaveCpuRank, const bool useSuperPodMode,
30 28 : DeviceMem expMem)
31 28 : : linkDummy_(nullptr),
32 28 : collectiveId_(collectiveId),
33 28 : userRank_(userRank),
34 28 : userRankSize_(userRankSize),
35 28 : rank_(rank),
36 28 : rankSize_(rankSize),
37 28 : paraVector_(paraVector),
38 56 : transportType_(rankSize, TransportType::TRANS_TYPE_RESERVED),
39 28 : deviceLogicId_(HCCL_DEFAULT_INITIAL_VALUE),
40 28 : devicePhyId_(INVALID_UINT),
41 28 : topoFlag_(topoFlag),
42 28 : tag_(tag),
43 56 : transportInfo_(rankSize),
44 56 : rankMap_(userRankSize, INVALID_VALUE_RANKID),
45 56 : userRankMap_(rankSize, INVALID_VALUE_RANKID),
46 28 : dispatcher_(dispatcher),
47 28 : notifyPool_(notifyPool),
48 28 : netDevCtxMap_(netDevCtxMap),
49 28 : exchanger_(exchanger),
50 28 : inputMem_(inputMem),
51 28 : outputMem_(outputMem),
52 28 : isUsedRdmaLevel0_(isUsedRdmaLevel0),
53 28 : dstInterServerMap_(),
54 28 : dstInterClientMap_(),
55 28 : dstIntraServerVec_(),
56 28 : dstIntraClientVec_(),
57 28 : linkThreads_(),
58 28 : threadsRapplyNum_(0),
59 28 : shmDev_(0),
60 28 : isAlltoAllCommMesh_(isAlltoAllCommMesh),
61 28 : nicDeployInner_(nicDeployInner),
62 28 : isNeedHeterogP2P_(false),
63 28 : useOneDoorbell_(useOneDoorbell),
64 28 : isAicpuModeEn_(isAicpuModeEn),
65 28 : subUserRankRoot_(rankRoot),
66 28 : isHaveCpuRank_(isHaveCpuRank),
67 28 : useSuperPodMode_(useSuperPodMode),
68 140 : expMem_(expMem)
69 28 : {}
70 :
71 41 : CommBase::~CommBase() { (void)DeInit(); }
72 :
73 28 : HcclResult CommBase::DeInit()
74 : {
75 28 : for (u32 index = 0; index < linkThreads_.size(); index++) {
76 0 : if (linkThreads_[index]) {
77 0 : if (linkThreads_[index]->joinable()) {
78 0 : HCCL_DEBUG("Joining Link Thread[%u]", index);
79 0 : linkThreads_[index]->join(); // 等待线程执行后释放资源
80 : }
81 :
82 0 : HcclResult ret = hrtResetDevice(deviceLogicId_); // 防止线程里面异常退出，在进程中reset
83 0 : CHK_PRT_RET(
84 : ret != HCCL_SUCCESS, HCCL_ERROR("[CommBase][DeInit]comm base reset device[%d] failed", deviceLogicId_),
85 : ret);
86 : }
87 : }
88 28 : linkThreads_.clear();
89 :
90 57 : for (u32 i = 0; i < transportInfo_.size(); i++) {
91 29 : if (transportInfo_[i]) { // 使用对应类型的port销毁
92 0 : CHK_RET(transportInfo_[i]->DeInit());
93 : }
94 : }
95 :
96 28 : return HCCL_SUCCESS;
97 : }
98 :
99 18 : HcclResult CommBase::Init()
100 : {
101 : // 获取rank->userrank以及userrank->rank的映射关系
102 18 : CHK_RET(SetRankMap());
103 :
104 18 : if (!IsGeneralServer()) {
105 : // 获取当前线程操作的设备ID
106 18 : CHK_RET(hrtGetDevice(&deviceLogicId_));
107 18 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId_), devicePhyId_));
108 : }
109 :
110 18 : intraSocketsMap_.insert(exchanger_.socketsMap.begin(), exchanger_.socketsMap.end());
111 :
112 : // 创建当前rank与其他rank之间的link(RDMA异步、TCP)
113 18 : CHK_RET(CreateLinks());
114 :
115 : // 校验当前rank与dst rank建链的链路有效性
116 18 : CHK_RET(CheckLinks());
117 :
118 : // task多线程并行下发，根据当前transport创建vtransport信息
119 18 : CHK_RET(CreateVirturalTransport());
120 :
121 18 : return HCCL_SUCCESS;
122 : }
123 :
124 1 : std::shared_ptr<Transport>& CommBase::GetTransportByRank(const u32 dstRank)
125 : {
126 1 : if (transportInfo_.size() <= dstRank) {
127 1 : HCCL_ERROR("[Get][TransportByRank]dstRank[%u] is bigger than link size[%llu]", dstRank, transportInfo_.size());
128 1 : return linkDummy_;
129 : }
130 :
131 0 : return transportInfo_[dstRank];
132 : }
133 :
134 6 : HcclResult CommBase::GetRankByUserRank(const u32 userRank, u32& rank) const
135 : {
136 6 : if (rankMap_.size() > userRank) {
137 4 : rank = rankMap_[userRank];
138 4 : if (rank == INVALID_VALUE_RANKID) {
139 2 : HCCL_INFO("This userRank[%u] is not in this sub communication. ", userRank);
140 2 : return HCCL_E_NOT_FOUND;
141 : }
142 2 : return HCCL_SUCCESS;
143 : }
144 :
145 2 : HCCL_ERROR("[Get][RankByUserRank]This userRank[%u] is invalid. ", userRank);
146 2 : rank = INVALID_VALUE_RANKID;
147 2 : return HCCL_E_PARA;
148 : }
149 :
150 6 : HcclResult CommBase::GetUserRankByRank(const u32 rank, u32& userRank) const
151 : {
152 6 : if (userRankMap_.size() > rank) {
153 2 : if (userRankMap_[rank] == INVALID_VALUE_RANKID) {
154 0 : HCCL_INFO("This rank[%u] is not in this sub communication.", rank);
155 0 : userRank = INVALID_VALUE_RANKID;
156 0 : return HCCL_E_NOT_FOUND;
157 : }
158 :
159 2 : userRank = userRankMap_[rank];
160 2 : return HCCL_SUCCESS;
161 : }
162 :
163 4 : HCCL_ERROR("[Get][UserRankByRank]This rank[%u] is invalid.", rank);
164 4 : userRank = INVALID_VALUE_RANKID;
165 4 : return HCCL_E_PARA;
166 : }
167 :
168 18 : HcclResult CommBase::CreateLinks()
169 : {
170 18 : HCCL_DEBUG("[CreateLinks] [comm_base] rankSize_[%u]", rankSize_);
171 18 : if (rankSize_ == HCCL_RANK_SIZE_EQ_ONE) {
172 18 : HCCL_INFO("comm base needn't to create links, rankSize_[%u].", rankSize_);
173 18 : return HCCL_SUCCESS;
174 : }
175 :
176 0 : CHK_RET(CalcLink());
177 : u32 threadsNum
178 0 : = dstInterClientMap_.size() + dstIntraClientVec_.size() + dstInterServerMap_.size() + dstIntraServerVec_.size();
179 0 : CHK_PRT_RET(
180 : (threadsNum == 0), HCCL_ERROR("[Create][Links]no link to create, threadsNum[%u]", threadsNum), HCCL_E_INTERNAL);
181 :
182 0 : linkThreads_.resize(threadsNum);
183 0 : HCCL_INFO(
184 : "comm base threads info:link threads size[%llu], dst inter client map size[%llu], "
185 : "dst intra client vec size[%llu], dst inter server map size[%llu], dst intra server vec size[%llu]",
186 : linkThreads_.size(), dstInterClientMap_.size(), dstIntraClientVec_.size(), dstInterServerMap_.size(),
187 : dstIntraServerVec_.size());
188 :
189 0 : CHK_RET(CreateExchangerNetwork());
190 :
191 0 : CHK_RET(CreateIntraLinks());
192 :
193 0 : CHK_RET(CreateInterLinks());
194 :
195 0 : bool check = (threadsRapplyNum_ != linkThreads_.size());
196 0 : CHK_PRT_RET(
197 : check,
198 : HCCL_ERROR(
199 : "[Create][Links]comm base rapply num[%u] is not equal to link threads[%llu]", threadsRapplyNum_,
200 : linkThreads_.size()),
201 : HCCL_E_INTERNAL);
202 :
203 0 : for (u32 index = 0; index < linkThreads_.size(); index++) {
204 0 : if (linkThreads_[index] == nullptr) {
205 0 : continue;
206 : }
207 0 : if (linkThreads_[index]->joinable()) {
208 0 : HCCL_DEBUG("Joining Link Thread[%u]", index);
209 0 : linkThreads_[index]->join(); // 等待线程执行完毕
210 : }
211 0 : if (!IsGeneralServer()) {
212 0 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
213 : }
214 : }
215 0 : linkThreads_.clear();
216 : // 建链结束立即释放socket资源（添加判断host网卡走的是TCP就不释放资源）
217 0 : if (pyhIdResourseSockets_.size()) {
218 0 : for (auto& iter : pyhIdResourseSockets_) {
219 0 : iter.second->DestroySockets(tag_);
220 : }
221 0 : } else if (!GetExternalInputHcclIsTcpMode() && interSocketManager_ != nullptr) {
222 : // 建链结束，关闭socket
223 0 : interSocketManager_->DestroySockets(tag_);
224 : }
225 0 : return HCCL_SUCCESS;
226 : }
227 :
228 0 : HcclResult CommBase::CalcLink() { return HCCL_SUCCESS; }
229 :
230 0 : u32 CommBase::GetSocketsPerLink() { return 1; }
231 :
232 2 : bool CommBase::NeedDataReceivedAck() { return false; }
233 :
234 : // 获取rank间的link type
235 1 : HcclResult CommBase::SetTransportType(const u32 dstRank)
236 : {
237 1 : LinkTypeInServer linkType = LinkTypeInServer::RESERVED_LINK_TYPE;
238 :
239 : // 适配910_93的RDMA+SIO ring,创建RDMA类型下的SIO连接
240 1 : if (linkType == LinkTypeInServer::SIO_TYPE && paraVector_[rank_].deviceType == DevType::DEV_TYPE_910_93) {
241 0 : transportType_[dstRank] = TransportType::TRANS_TYPE_P2P;
242 1 : } else if (paraVector_[rank_].serverId == paraVector_[dstRank].serverId) { // 判断是否在同一个server
243 : // Server内判断是否使用rdma
244 0 : if (isUsedRdmaLevel0_ || isAlltoAllCommMesh_
245 0 : || (paraVector_[rank_].deviceType != DevType::DEV_TYPE_310P3
246 0 : && (paraVector_[rank_].devicePhyId / HCCL_AISERVER_DEVICE_NUM
247 0 : != paraVector_[dstRank].devicePhyId / HCCL_AISERVER_DEVICE_NUM))) {
248 0 : transportType_[dstRank] = TransportType::TRANS_TYPE_IBV_EXP;
249 : } else {
250 0 : transportType_[dstRank] = TransportType::TRANS_TYPE_P2P;
251 : }
252 : } else { // server间
253 1 : if (IsSupportInterHccs(dstRank)) {
254 : // 超节点内节点间走HCCS通信
255 0 : transportType_[dstRank] = TransportType::TRANS_TYPE_P2P;
256 : } else {
257 1 : transportType_[dstRank] = TransportType::TRANS_TYPE_IBV_EXP;
258 : }
259 : }
260 :
261 1 : HCCL_INFO("SetTransportType: dstRank[%u] transport_type[%d]", dstRank, transportType_[dstRank]);
262 1 : return HCCL_SUCCESS;
263 : }
264 :
265 0 : HcclResult CommBase::RunTemplateAlg(const std::unique_ptr<AlgTemplateBase>& tempAlg)
266 : {
267 0 : HcclResult ret = tempAlg->RunAsync(Rank(), RankSize(), transportInfo_);
268 0 : CHK_PRT_RET(ret == HCCL_E_AGAIN, HCCL_WARNING("[Run][AlgTemplateBase]group has been destroyed. Break!"), ret);
269 0 : CHK_PRT_RET(
270 : ret != HCCL_SUCCESS,
271 : HCCL_ERROR(
272 : "[Run][AlgTemplateBase]comm base run tempAlg "
273 : "rank[%u] rank size[%u] failed",
274 : rank_, rankSize_),
275 : ret);
276 0 : return HCCL_SUCCESS;
277 : }
278 :
279 0 : HcclResult CommBase::RunTemplateAlgStaged(const std::unique_ptr<AlgTemplateBase>& tempAlg, const RunStage& stage)
280 : {
281 0 : HcclResult ret = tempAlg->RunAsyncStaged(Rank(), RankSize(), transportInfo_, stage);
282 0 : CHK_PRT_RET(
283 : ret != HCCL_SUCCESS,
284 : HCCL_ERROR(
285 : "[Run][RunTemplateAlgStaged]comm base run tempAlg staged "
286 : "rank[%u] rank size[%u] failed",
287 : rank_, rankSize_),
288 : ret);
289 0 : return HCCL_SUCCESS;
290 : }
291 :
292 18 : HcclResult CommBase::SetRankMap()
293 : {
294 : // 参数有效性校验
295 18 : if ((userRankSize_ <= userRank_) || (rankSize_ <= rank_)) {
296 0 : HCCL_ERROR(
297 : "[Set][RankMap]invalid:userRankSize_[%u] userRank_[%u] rankSize_[%u] rank_[%u].", userRankSize_, userRank_,
298 : rankSize_, rank_);
299 0 : return HCCL_E_PARA;
300 : }
301 :
302 36 : for (u32 index = 0; index < rankSize_; index++) {
303 18 : userRankMap_[index] = paraVector_[index].userRank;
304 :
305 18 : if (userRankSize_ > userRankMap_[index]) {
306 18 : rankMap_[userRankMap_[index]] = index;
307 : }
308 18 : HCCL_INFO("userRankMap: [%u] -> [%u]", index, paraVector_[index].worldRank);
309 : }
310 :
311 18 : return HCCL_SUCCESS;
312 : }
313 :
314 18 : HcclResult CommBase::CheckLinks() const
315 : {
316 36 : for (u32 index = 0; index < transportType_.size(); index++) {
317 18 : bool check = (transportInfo_.size() <= index);
318 18 : CHK_PRT_RET(
319 : check, HCCL_ERROR("[Check][Links]index[%u] is bigger than link size[%llu]", index, transportInfo_.size()),
320 : HCCL_E_INTERNAL);
321 18 : if ((transportType_[index] != TransportType::TRANS_TYPE_RESERVED) && !transportInfo_[index]) {
322 0 : HCCL_ERROR(
323 : "[Check][Links]there is no effective link(type[%d]) between rank[%u] and dst rank[%u]!",
324 : transportType_[index], rank_, index);
325 0 : return HCCL_E_NOT_FOUND;
326 : }
327 : }
328 :
329 18 : return HCCL_SUCCESS;
330 : }
331 :
332 : // 只有节点内，采用虚拟网卡时，才会进入该函数，有且仅有一个nic_ip
333 0 : HcclResult CommBase::CreateIntraLinks()
334 : {
335 0 : HcclUs startut = TIME_NOW();
336 0 : HcclResult ret = HCCL_SUCCESS;
337 :
338 0 : auto socketsMap = intraSocketsMap_;
339 0 : HCCL_DEBUG("[Create][IntraLinks] dstIntraServerVec size[%u].", dstIntraServerVec_.size());
340 0 : for (auto& rank : dstIntraServerVec_) {
341 0 : HCCL_DEBUG("[Create][IntraLinks] localrank[%u] remoterank[%u].", userRank_, rank);
342 : // 与当前Inter的Socket不同, 在Intra Socket创建时, 使用的userRank, 所以这里需要使用 userRank 为 key
343 0 : auto item = socketsMap.find(paraVector_[rank].userRank);
344 0 : if (item != socketsMap.end()) {
345 0 : ret = CreateIntraThread(CLIENT_ROLE_SOCKET, rank, item->second);
346 : } else {
347 0 : HCCL_INFO("[Create][IntraLinks] remoterank[%u] socket item not find.", rank);
348 : // 异构场景下, 当前上层 CreateCommP2PAsync 并不会创建 IntraExchanger, 后继的
349 : // TransportHeterogP2P 场景使用原有的 Socket 逻辑, 所以在这里没有找到 socket item 时,
350 : // 使用一个空的 sockets 作为参数, 创建后继处理线程.
351 0 : std::vector<std::shared_ptr<HcclSocket>> sockets;
352 0 : ret = CreateIntraThread(CLIENT_ROLE_SOCKET, rank, sockets);
353 0 : }
354 0 : CHK_PRT_RET(
355 : ret != HCCL_SUCCESS, HCCL_ERROR("[Create][IntraLinks] create intra thread failed, socket role is client"),
356 : ret);
357 : }
358 :
359 0 : HCCL_DEBUG("[Create][IntraLinks] dstIntraClientVec size[%u].", dstIntraClientVec_.size());
360 0 : for (auto& rank : dstIntraClientVec_) {
361 0 : HCCL_DEBUG("[Create][IntraLinks] localrank[%u] remoterank[%u].", userRank_, rank);
362 0 : auto item = socketsMap.find(paraVector_[rank].userRank);
363 0 : if (item != socketsMap.end()) {
364 0 : ret = CreateIntraThread(SERVER_ROLE_SOCKET, rank, item->second);
365 : } else {
366 0 : HCCL_INFO("[Create][IntraLinks] remoterank[%u] socket item not find.", rank);
367 : // 异构场景下, 当前上层 CreateCommP2PAsync 并不会创建 IntraExchanger, 后继的
368 : // TransportHeterogP2P 场景使用原有的 Socket 逻辑, 所以在这里没有找到 socket item 时,
369 : // 使用一个空的 sockets 作为参数, 创建后继处理线程.
370 0 : std::vector<std::shared_ptr<HcclSocket>> sockets;
371 0 : ret = CreateIntraThread(SERVER_ROLE_SOCKET, rank, sockets);
372 0 : }
373 0 : CHK_PRT_RET(
374 : ret != HCCL_SUCCESS, HCCL_ERROR("[Create][IntraLinks] create intra thread failed, socket role is server"),
375 : ret);
376 : }
377 :
378 0 : HCCL_DEBUG("[Create][IntraLinks] create intra link used time:%lld us.", DURATION_US(TIME_NOW() - startut));
379 0 : return ret;
380 0 : }
381 :
382 : HcclResult
383 0 : CommBase::CreateIntraThread(const u32 role, u32 dstRank, const std::vector<std::shared_ptr<HcclSocket>>& sockets)
384 : {
385 0 : if (threadsRapplyNum_ >= linkThreads_.size()) {
386 0 : HCCL_ERROR(
387 : "[Create][InterThread] threadsRapplyNum_[%u] is bigger than link threads size[%llu] ", threadsRapplyNum_,
388 : linkThreads_.size());
389 0 : return HCCL_E_INTERNAL;
390 : }
391 :
392 : // 线程命名，TraL_ 代表Intra Link
393 0 : std::string threadStr = "HcclTraL_" + std::to_string(threadsRapplyNum_);
394 :
395 : // 创建新线程前更新一下最新的workflowMode
396 0 : workflowMode_ = GetWorkflowMode();
397 0 : if (role == SERVER_ROLE_SOCKET) {
398 0 : linkThreads_[threadsRapplyNum_].reset(new (std::nothrow) std::thread(
399 0 : &CommBase::CreateDestLink, this, hrtErrMGetErrorContextPub(), MachineType::MACHINE_SERVER_TYPE,
400 0 : paraVector_[rank_].serverId, dstRank, threadStr, sockets));
401 : }
402 :
403 0 : HCCL_DEBUG("[CommBase][CreateIntraThread]role is %u", role);
404 0 : if (role == CLIENT_ROLE_SOCKET) {
405 0 : linkThreads_[threadsRapplyNum_].reset(new (std::nothrow) std::thread(
406 0 : &CommBase::CreateDestLink, this, hrtErrMGetErrorContextPub(), MachineType::MACHINE_CLIENT_TYPE,
407 0 : paraVector_[rank_].serverId, dstRank, threadStr, sockets));
408 : }
409 :
410 0 : if (!linkThreads_[threadsRapplyNum_]) {
411 0 : HCCL_ERROR("[Create][IntraThread] link threads[%u] reset failed.", threadsRapplyNum_);
412 0 : return HCCL_E_INTERNAL;
413 : }
414 0 : threadsRapplyNum_++;
415 :
416 0 : HCCL_DEBUG(
417 : "[Create][IntraThread] role[%u], dstRank[%u], sockets size[%u], threadsRapplyNum[%u]", role, dstRank,
418 : sockets.size(), threadsRapplyNum_);
419 0 : return HCCL_SUCCESS;
420 0 : }
421 :
422 0 : void CommBase::PrintCreateInterLinksInfo()
423 : {
424 0 : HCCL_RUN_INFO(
425 : "[PrintCreateInterLinksInfo] dstInterServerMap size[%llu], dstInterClientMap size[%llu]",
426 : dstInterServerMap_.size(), dstInterClientMap_.size());
427 :
428 : // 维护建链输出的信息
429 0 : std::string outLogInfo = "";
430 0 : for (auto iter = dstInterServerMap_.begin(); iter != dstInterServerMap_.end(); iter++) {
431 0 : outLogInfo.append(std::to_string(paraVector_[iter->first].userRank));
432 0 : outLogInfo.append("/");
433 0 : outLogInfo.append(paraVector_[iter->first].serverId);
434 0 : outLogInfo.append("/");
435 0 : outLogInfo.append(std::to_string(paraVector_[iter->first].devicePhyId));
436 0 : outLogInfo.append("; ");
437 : }
438 :
439 0 : for (auto iter = dstInterClientMap_.begin(); iter != dstInterClientMap_.end(); iter++) {
440 0 : outLogInfo.append(std::to_string(paraVector_[iter->first].userRank));
441 0 : outLogInfo.append("/");
442 0 : outLogInfo.append(paraVector_[iter->first].serverId);
443 0 : outLogInfo.append("/");
444 0 : outLogInfo.append(std::to_string(paraVector_[iter->first].devicePhyId));
445 0 : outLogInfo.append("; ");
446 : }
447 :
448 0 : HCCL_RUN_INFO(
449 : "serverInterConnectInfo:tag[%s], userRank/serverIp/devicePhyId:[%u/%s/%d], connectRankInfo[%s]", tag_.c_str(),
450 : userRank_, paraVector_[rank_].serverId.c_str(), paraVector_[rank_].devicePhyId, outLogInfo.c_str());
451 0 : }
452 :
453 0 : HcclResult CommBase::CreateInterLinks()
454 : {
455 0 : interSocketManager_.reset(new (std::nothrow)
456 0 : HcclSocketManager(nicDeployInner_, deviceLogicId_, devicePhyId_, userRank_));
457 0 : CHK_PTR_NULL(interSocketManager_);
458 :
459 0 : if (dstInterServerMap_.size() + dstInterClientMap_.size() == 0) {
460 0 : HCCL_DEBUG("[Create][InterLinks] do not need create links.");
461 0 : return HCCL_SUCCESS;
462 : }
463 :
464 0 : PrintCreateInterLinksInfo();
465 :
466 0 : HcclUs startut = TIME_NOW();
467 0 : HcclResult ret = HCCL_SUCCESS;
468 0 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> serverSocketsMap;
469 0 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> clientSocketsMap;
470 0 : ret = interSocketManager_->CreateSockets(
471 0 : tag_, true, netDevCtxMap_[paraVector_[rank_].nicIp[0]], dstInterServerMap_, dstInterClientMap_,
472 : serverSocketsMap, clientSocketsMap);
473 0 : CHK_PRT_RET(
474 : ret != HCCL_SUCCESS, HCCL_ERROR("[Create][InterLinks] socket manager create connections failed, ret[%u]", ret),
475 : ret);
476 :
477 0 : for (auto& sockets : clientSocketsMap) {
478 0 : ret = CreateInterThread(CLIENT_ROLE_SOCKET, sockets.first, sockets.second);
479 0 : CHK_PRT_RET(
480 : ret != HCCL_SUCCESS,
481 : HCCL_ERROR("[Create][InterLinks] create inter thread failed, socket role[CLIENT_ROLE_SOCKET] "), ret);
482 : }
483 0 : HCCL_DEBUG("[CommBase][CreateInterLinks]create inter thread success");
484 0 : for (auto& sockets : serverSocketsMap) {
485 0 : ret = CreateInterThread(SERVER_ROLE_SOCKET, sockets.first, sockets.second);
486 0 : CHK_PRT_RET(
487 : ret != HCCL_SUCCESS,
488 : HCCL_ERROR("[Create][InterLinks] create inter thread failed, socket role[SERVER_ROLE_SOCKET] "), ret);
489 : }
490 :
491 0 : HCCL_DEBUG("[Create][InterLinks] create inter link used time:%lld us", DURATION_US(TIME_NOW() - startut));
492 0 : return ret;
493 0 : }
494 :
495 : HcclResult
496 0 : CommBase::CreateInterThread(const u32 role, u32 dstRank, const std::vector<std::shared_ptr<HcclSocket>>& sockets)
497 : {
498 0 : if (sockets.empty()) {
499 0 : HCCL_ERROR("[Create][InterThread] create inter link failed, rank's sockets is empty");
500 0 : return HCCL_E_INTERNAL;
501 : }
502 :
503 0 : if (threadsRapplyNum_ >= linkThreads_.size()) {
504 0 : HCCL_ERROR(
505 : "[Create][InterThread] threadsRapplyNum_[%u] is bigger than link threads size[%llu] ", threadsRapplyNum_,
506 : linkThreads_.size());
507 0 : return HCCL_E_INTERNAL;
508 : }
509 :
510 : // 线程命名，TerL代表Inter Link
511 0 : std::string threadStr = "HcclTerL_" + std::to_string(threadsRapplyNum_);
512 :
513 : // 创建新线程前更新一下最新的workflowMode
514 0 : workflowMode_ = GetWorkflowMode();
515 0 : if (role == SERVER_ROLE_SOCKET) {
516 0 : linkThreads_[threadsRapplyNum_].reset(new (std::nothrow) std::thread(
517 0 : &CommBase::CreateDestLink, this, hrtErrMGetErrorContextPub(), MachineType::MACHINE_SERVER_TYPE,
518 0 : paraVector_[rank_].serverId, dstRank, threadStr, sockets));
519 : }
520 :
521 0 : if (role == CLIENT_ROLE_SOCKET) {
522 0 : linkThreads_[threadsRapplyNum_].reset(new (std::nothrow) std::thread(
523 0 : &CommBase::CreateDestLink, this, hrtErrMGetErrorContextPub(), MachineType::MACHINE_CLIENT_TYPE,
524 0 : paraVector_[rank_].serverId, dstRank, threadStr, sockets));
525 : }
526 :
527 0 : if (!linkThreads_[threadsRapplyNum_]) {
528 0 : HCCL_ERROR("[Create][InterThread] link threads[%u] reset failed.", threadsRapplyNum_);
529 0 : return HCCL_E_INTERNAL;
530 : }
531 0 : threadsRapplyNum_++;
532 :
533 0 : HCCL_DEBUG(
534 : "[Create][InterThread] role[%u], dstRank[%u], sockets size[%u], threadsRapplyNum[%u]", role, dstRank,
535 : sockets.size(), threadsRapplyNum_);
536 :
537 0 : return HCCL_SUCCESS;
538 0 : }
539 :
540 0 : u32 CommBase::GetInterRemotePort(s32 devicePhyId, u32 dstUserRank)
541 : {
542 0 : if (isUseRankPort_ && dstUserRank < ranksPort_.size() && ranksPort_[dstUserRank] != HCCL_INVALID_PORT) {
543 0 : HCCL_INFO("[GetInterRemotePort] port[%u] from ranks port", ranksPort_[dstUserRank]);
544 0 : return ranksPort_[dstUserRank];
545 0 : } else if (!isUseRankPort_ && !Is310PDevice()) {
546 0 : HCCL_INFO("[GetInterRemotePort] port[%u]", HETEROG_CCL_PORT);
547 0 : return HETEROG_CCL_PORT;
548 0 : } else if (GetExternalInputHcclIfBasePort() == HCCL_INVALID_PORT) {
549 0 : return (devicePhyId + HOST_PARA_BASE_PORT);
550 : } else {
551 0 : return (devicePhyId + GetExternalInputHcclIfBasePort() + HCCL_AISERVER_DEVICE_NUM);
552 : }
553 : }
554 :
555 0 : HcclResult CommBase::CalcLinksNum(const MachineType machineType, const u32 dstRank)
556 : {
557 0 : bool check = (paraVector_.size() <= dstRank) || (paraVector_.size() <= rank_);
558 0 : CHK_PRT_RET(
559 : check,
560 : HCCL_ERROR(
561 : "[Calc][LinksNum]para check failed, para vector size[%llu], dstRank[%u], rank[%u] ", paraVector_.size(),
562 : dstRank, rank_),
563 : HCCL_E_INTERNAL);
564 : // 节点间或者是节点内采用RDMA通信的, 放至dst_inter_client_map_,采用rdma建链
565 : bool isInterRdma
566 0 : = paraVector_[rank_].serverId != paraVector_[dstRank].serverId || isUsedRdmaLevel0_ || isAlltoAllCommMesh_;
567 :
568 0 : bool isInterHccs = IsSupportInterHccs(dstRank);
569 :
570 0 : HCCL_DEBUG(
571 : "[Calc][LinksNum]rank[%u], dstRank[%u], isInterRdma[%d], isInterHccs[%d], machineType[%d]", rank_, dstRank,
572 : isInterRdma, isInterHccs, machineType);
573 :
574 0 : auto dstRankInfo = paraVector_[dstRank];
575 0 : if (machineType == MachineType::MACHINE_SERVER_TYPE) {
576 0 : CHK_RET(MakeClientInfo(dstRank, dstRankInfo, isInterRdma, isInterHccs));
577 : }
578 :
579 0 : if (machineType == MachineType::MACHINE_CLIENT_TYPE) {
580 0 : CHK_RET(MakeServerInfo(dstRank, dstRankInfo, isInterRdma, isInterHccs));
581 : }
582 :
583 0 : return HCCL_SUCCESS;
584 0 : }
585 :
586 0 : HcclResult CommBase::MakeClientInfo(const u32 dstRank, RankInfo& dstRankInfo, bool isInterRdma, bool isInterHccs)
587 : {
588 0 : if (isInterRdma && !isInterHccs) {
589 0 : HcclRankLinkInfo tempLinkInfo{};
590 0 : tempLinkInfo.userRank = dstRankInfo.userRank;
591 0 : tempLinkInfo.devicePhyId = dstRankInfo.devicePhyId;
592 :
593 0 : tempLinkInfo.ip = dstRankInfo.nicIp[0];
594 0 : tempLinkInfo.port = GetInterRemotePort(tempLinkInfo.devicePhyId, dstRankInfo.userRank);
595 0 : tempLinkInfo.socketsPerLink = GetSocketsPerLink();
596 :
597 0 : auto iter = dstInterClientMap_.find(dstRank);
598 0 : bool check = (iter != dstInterClientMap_.end());
599 0 : CHK_PRT_RET(
600 : check, HCCL_ERROR("[Make][ClientInfo]dstRank[%u] already exists in dst inter client map. ", dstRank),
601 : HCCL_E_PARA);
602 0 : dstInterClientMap_.insert(std::make_pair(dstRank, tempLinkInfo));
603 0 : } else {
604 0 : dstIntraClientVec_.push_back(dstRank);
605 : }
606 0 : return HCCL_SUCCESS;
607 : }
608 :
609 0 : HcclResult CommBase::MakeServerInfo(const u32 dstRank, RankInfo& dstRankInfo, bool isInterRdma, bool isInterHccs)
610 : {
611 : // 节点间或者是节点内采用RDMA通信的，放至dst_inter_client_map_,采用rdma建链
612 0 : if (isInterRdma && !isInterHccs) {
613 0 : HcclRankLinkInfo tempLinkInfo{};
614 0 : tempLinkInfo.userRank = dstRankInfo.userRank;
615 0 : tempLinkInfo.devicePhyId = dstRankInfo.devicePhyId;
616 :
617 0 : HCCL_INFO(
618 : "dstRank = %u, useRank = %u, ip = %s", dstRank, dstRankInfo.userRank,
619 : dstRankInfo.nicIp[0].GetReadableAddress());
620 :
621 0 : tempLinkInfo.ip = dstRankInfo.nicIp[0];
622 0 : tempLinkInfo.port = GetInterRemotePort(tempLinkInfo.devicePhyId, dstRankInfo.userRank);
623 0 : tempLinkInfo.socketsPerLink = GetSocketsPerLink();
624 :
625 0 : auto iter = dstInterServerMap_.find(dstRank);
626 0 : bool check = (iter != dstInterServerMap_.end());
627 0 : CHK_PRT_RET(
628 : check, HCCL_ERROR("[Make][ServerInfo]dstRank[%u] already exists in dst inter server map", dstRank),
629 : HCCL_E_PARA);
630 0 : dstInterServerMap_.insert(std::make_pair(dstRank, tempLinkInfo));
631 0 : } else {
632 0 : dstIntraServerVec_.push_back(dstRank);
633 : }
634 0 : return HCCL_SUCCESS;
635 : }
636 :
637 1 : HcclResult CommBase::CreateDestLink(
638 : const ErrContextPub& error_context, const MachineType machineType, const std::string& serverId, const u32 dstRank,
639 : const std::string& threadStr, const std::vector<std::shared_ptr<HcclSocket>>& sockets)
640 : {
641 1 : hrtErrMSetErrorContextPub(error_context);
642 : // 给当前线程添加名字
643 1 : SetThreadName(threadStr);
644 1 : if (!IsGeneralServer()) {
645 1 : CHK_RET(hrtSetDevice(deviceLogicId_));
646 1 : SetWorkflowMode(workflowMode_); // 新的线程，更新workflowMode
647 : }
648 :
649 2 : bool check = (paraVector_.size() <= dstRank) || (paraVector_.size() <= rank_) || (transportInfo_.size() <= dstRank)
650 2 : || (transportType_.size() <= dstRank);
651 1 : CHK_PRT_RET(
652 : check,
653 : HCCL_ERROR(
654 : "[Create][DestLink]paraCheck failed, paraVector size[%llu], linkInfo size[%llu], "
655 : "linkType size[%llu], dstRank[%u], rank[%u] ",
656 : paraVector_.size(), transportInfo_.size(), transportType_.size(), dstRank, rank_),
657 : HCCL_E_INTERNAL);
658 :
659 1 : MachinePara machinePara;
660 1 : CHK_RET(SetMachinePara(machineType, serverId, dstRank, sockets, machinePara));
661 1 : HCCL_INFO(
662 : "[creakLink para]rank[%u]-localUserrank[%u]-localIpAddr[%s], linkMode[%d] "
663 : "dst_rank[%u]-remoteUserrank[%u]-remote_ip_addr[%s], machineType[%d], serverId[%s], nicDeploy[%d] ",
664 : rank_, paraVector_[rank_].worldRank, paraVector_[rank_].serverId.c_str(), machinePara.linkMode, dstRank,
665 : paraVector_[dstRank].worldRank, paraVector_[dstRank].serverId.c_str(), machinePara.machineType,
666 : machinePara.serverId.c_str(), machinePara.nicDeploy);
667 :
668 : // transport初始化
669 1 : HcclResult ret = TransportInit(dstRank, machinePara);
670 1 : if (ret != HCCL_SUCCESS) {
671 1 : transportInfo_[dstRank] = nullptr;
672 1 : if (ret == HCCL_E_MEMORY) {
673 : std::string err_str
674 : = "[Create][DestLink]Transport init error! IPC memory allocation failed due to "
675 : "possible memory limit exceeded. Suggested solution: Use 3TB / (ranksize * 2) as the upper limit of "
676 1 : "HCCL_BUFFSIZE.";
677 1 : HCCL_ERROR("%s", err_str.c_str());
678 1 : }
679 : const std::string CREATE_LINK_ERR
680 2 : = "[Create][DestLink]Create Dest error! creakLink para:rank[" + std::to_string(rank_) + "]-localUserrank["
681 4 : + std::to_string(paraVector_[rank_].worldRank) + "]-localIpAddr[" + paraVector_[rank_].serverId.c_str()
682 4 : + "], dst_rank[" + std::to_string(dstRank) + "]-remoteUserrank["
683 4 : + std::to_string(paraVector_[dstRank].worldRank) + "]-remote_ip_addr["
684 2 : + paraVector_[dstRank].serverId.c_str() + "]";
685 :
686 1 : HCCL_ERROR(
687 : "[Create][DestLink]Transport init error! creakLink para:rank[%u]-localUserrank[%u]-localIpAddr[%s], "
688 : "dst_rank[%u]-remoteUserrank[%u]-remote_ip_addr[%s], machineType[%d], serverId[%s], linkMode[%d], "
689 : "shmDev_[%u], tag[%s]",
690 : rank_, paraVector_[rank_].worldRank, paraVector_[rank_].serverId.c_str(), dstRank,
691 : paraVector_[dstRank].worldRank, paraVector_[dstRank].serverId.c_str(), machinePara.machineType,
692 : machinePara.serverId.c_str(), machinePara.linkMode, shmDev_, machinePara.tag.c_str());
693 1 : return ret;
694 1 : }
695 0 : HCCL_INFO(
696 : "[creakLink success]:rank[%u]-localUserrank[%u]-localIpAddr[%s], "
697 : "dst_rank[%u]-remoteUserrank[%u]-remote_ip_addr[%s], transportType_[%d], tag[%s]",
698 : rank_, paraVector_[rank_].worldRank, paraVector_[rank_].serverId.c_str(), dstRank,
699 : paraVector_[dstRank].worldRank, paraVector_[dstRank].serverId.c_str(), transportType_[dstRank],
700 : machinePara.tag.c_str());
701 :
702 0 : return HCCL_SUCCESS;
703 1 : }
704 :
705 2 : void CommBase::SetTransportParam(TransportPara& para, [[maybe_unused]] MachinePara& machinePara)
706 : {
707 2 : std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
708 2 : para.isRootRank = subUserRankRoot_ == rank_ ? true : false;
709 2 : para.timeout = kdefaultTimeout;
710 2 : para.virtualFlag = false;
711 2 : }
712 :
713 2 : HcclResult CommBase::TransportInit(const u32 dstRank, MachinePara& machinePara)
714 : {
715 2 : CHK_PRT_RET(
716 : dstRank >= transportInfo_.size(),
717 : HCCL_ERROR("[TransportQuerry] Transport[%u] is invalid, should init before query it.", dstRank), HCCL_E_PARA);
718 : // 实例化TransportBase
719 2 : CHK_RET(SetTransportType(dstRank));
720 2 : TransportPara para{};
721 2 : SetTransportParam(para, machinePara);
722 :
723 2 : TransportType type = transportType_[dstRank];
724 2 : if (type == TransportType::TRANS_TYPE_P2P) {
725 0 : transportInfo_[dstRank].reset(new (std::nothrow) Transport(type, para, dispatcher_, notifyPool_, machinePara));
726 2 : } else if (type == TransportType::TRANS_TYPE_IBV_EXP) {
727 0 : transportInfo_[dstRank].reset(new (std::nothrow) Transport(type, para, dispatcher_, notifyPool_, machinePara));
728 : } else {
729 2 : HCCL_ERROR("[Init][Transport]not supported transport type");
730 2 : return HCCL_E_NOT_SUPPORT;
731 : }
732 :
733 0 : CHK_PRT_RET(!transportInfo_[dstRank], HCCL_ERROR("[Init][Transport]In create link, new link failed"), HCCL_E_PTR);
734 :
735 0 : if (useOneDoorbell_) {
736 0 : transportInfo_[dstRank]->EnableUseOneDoorbell();
737 : }
738 :
739 0 : CHK_RET(transportInfo_[dstRank]->Init());
740 :
741 0 : CHK_RET(CheckExchangeInfo(transportInfo_[dstRank], machinePara.localDeviceId));
742 :
743 0 : return HCCL_SUCCESS;
744 : }
745 :
746 2 : HcclResult CommBase::SetMachinePara(
747 : MachineType machineType, const std::string& serverId, u32 dstRank,
748 : const std::vector<std::shared_ptr<HcclSocket>>& socketList, MachinePara& machinePara)
749 : {
750 2 : SetMachineLinkMode(machinePara);
751 2 : HCCL_INFO("[Set][MachinePara]rankSize %u, linkMode %d", rankSize_, machinePara.linkMode);
752 :
753 2 : machinePara.machineType = machineType;
754 2 : machinePara.serverId = serverId;
755 2 : machinePara.localIpAddr = paraVector_[rank_].nicIp[0];
756 2 : machinePara.remoteIpAddr = paraVector_[dstRank].nicIp[0];
757 2 : machinePara.localUserrank = paraVector_[rank_].userRank;
758 2 : machinePara.remoteUserrank = paraVector_[dstRank].userRank;
759 2 : machinePara.localWorldRank = paraVector_[rank_].worldRank;
760 2 : machinePara.remoteWorldRank = paraVector_[dstRank].worldRank;
761 2 : machinePara.collectiveId = collectiveId_;
762 2 : machinePara.localDeviceId = paraVector_[rank_].devicePhyId;
763 2 : machinePara.remoteDeviceId = paraVector_[dstRank].devicePhyId;
764 2 : machinePara.deviceType = static_cast<DevType>(paraVector_[dstRank].deviceType);
765 2 : machinePara.inputMem = inputMem_;
766 2 : machinePara.outputMem = outputMem_;
767 2 : if (expMem_.ptr() != nullptr) {
768 0 : machinePara.mem.push_back(expMem_);
769 : } else {
770 2 : machinePara.mem.clear();
771 : }
772 2 : machinePara.linkAttribute = 0x03; /* 0x03同时支持目的端和源端发起 */
773 2 : machinePara.tag = tag_;
774 :
775 : // MoE算子优化，MC2 多机场景使用普通QP模式
776 2 : const std::string& suffix = HCCL_MC2_MULTISERVER_SUFFIX;
777 2 : if (tag_.size() > suffix.size() && tag_.compare(tag_.size() - suffix.size(), suffix.size(), suffix) == 0) {
778 1 : bool isSupportNormalQP{false};
779 1 : CHK_RET(IsSupportAicpuNormalQP(paraVector_[rank_].devicePhyId, isSupportNormalQP));
780 1 : if (isSupportNormalQP) {
781 1 : HCCL_INFO("[Set][MachinePara] Set machinePara.qpMode to [NORMAL]");
782 1 : machinePara.qpMode = QPMode::NORMAL;
783 : }
784 : }
785 :
786 : // 把原来的两层vector变成一层, 方便后继调用
787 2 : for (u32 i = 0; i < socketList.size(); i++) {
788 0 : machinePara.sockets.push_back(socketList[i]);
789 : }
790 : u64 rankConsistentDataLength
791 2 : = RankConsistentcyChecker::GetInstance(machinePara.localDeviceId).GetRankConsistentDataLength();
792 2 : machinePara.exchangeInfo.resize(rankConsistentDataLength);
793 2 : CHK_RET(RankConsistentcyChecker::GetInstance(machinePara.localDeviceId)
794 : .GetCheckFrame(&machinePara.exchangeInfo[0], rankConsistentDataLength, tag_));
795 2 : machinePara.supportDataReceivedAck = NeedDataReceivedAck();
796 2 : machinePara.nicDeploy = nicDeployInner_;
797 2 : machinePara.localSocketPort = paraVector_[rank_].hostPort;
798 2 : machinePara.remoteSocketPort = paraVector_[dstRank].hostPort;
799 2 : machinePara.isAicpuModeEn = isAicpuModeEn_;
800 2 : machinePara.deviceLogicId = deviceLogicId_;
801 2 : machinePara.srcPorts = std::vector<std::uint16_t>(1, 0); /* 默认填充一个元素，0代表默认不配置 */
802 2 : return HCCL_SUCCESS;
803 : }
804 :
805 18 : HcclResult CommBase::CreateVirturalTransport()
806 : {
807 18 : MachinePara machinePara;
808 18 : std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
809 :
810 18 : vTransportInfo_.resize(transportInfo_.size());
811 36 : for (u32 i = 0; i < transportInfo_.size(); i++) {
812 18 : TransportPara para{};
813 18 : para.virtualFlag = true;
814 18 : para.timeout = kdefaultTimeout;
815 18 : para.index = i;
816 36 : vTransportInfo_[i].reset(new (std::nothrow) Transport(
817 36 : TransportType::TRANS_TYPE_RESERVED, para, dispatcher_, notifyPool_, machinePara));
818 :
819 18 : CHK_PRT_RET(
820 : !vTransportInfo_[i], HCCL_ERROR("[CreateVirturalTransport]In create link, new link failed"), HCCL_E_PTR);
821 : }
822 :
823 18 : return HCCL_SUCCESS;
824 18 : }
825 :
826 0 : std::shared_ptr<Transport>& CommBase::GetTrasportInfoByVTransportInfoIndex(u32 index)
827 : {
828 0 : if (vTransportInfo_.size() <= index) {
829 0 : HCCL_ERROR(
830 : "[GetTrasportInfoByVTransportInfoIndex]index[%u] is bigger than vlink size[%llu]", index,
831 : vTransportInfo_.size());
832 0 : return linkDummy_;
833 : }
834 :
835 0 : if (transportInfo_.size() <= index) {
836 0 : HCCL_ERROR(
837 : "[GetTrasportInfoByVTransportInfoIndex]index[%u] is bigger than link size[%llu]", index,
838 : transportInfo_.size());
839 0 : return linkDummy_;
840 : }
841 0 : return transportInfo_[index];
842 : }
843 0 : HcclResult CommBase::BuildAsync(u32& status)
844 : {
845 0 : transportStatus_.resize(rankSize_, 1);
846 0 : checkStatus_.resize(rankSize_, false);
847 :
848 : // 获取rank->userrank以及userrank->rank的映射关系
849 0 : CHK_RET(SetRankMap());
850 :
851 : // 获取当前线程操作的设备ID
852 0 : deviceLogicId_ = 0;
853 0 : if (paraVector_[rank_].devicePhyId != HOST_DEVICE_ID) {
854 0 : CHK_RET(hrtGetDevice(&deviceLogicId_));
855 : }
856 :
857 0 : if (rankSize_ == HCCL_RANK_SIZE_EQ_ONE) {
858 0 : HCCL_INFO("comm base needn't to create links, rankSize_[%u].", rankSize_);
859 0 : status = 0;
860 0 : return HCCL_SUCCESS;
861 : }
862 0 : CHK_RET(CalcLink());
863 :
864 : // 当前rank作为client端角色
865 0 : u32 dstIntraServerNum = dstIntraServerVec_.size();
866 0 : std::vector<std::shared_ptr<HcclSocket>> sockets;
867 0 : for (u32 intraIndex = 0; intraIndex < dstIntraServerNum; intraIndex++) {
868 0 : HcclResult ret = TransportBuildAsync(
869 0 : MachineType::MACHINE_CLIENT_TYPE, paraVector_[rank_].serverId, dstIntraServerVec_[intraIndex], sockets,
870 0 : transportStatus_[dstIntraServerVec_[intraIndex]]);
871 0 : CHK_PRT_RET(
872 : ret,
873 : HCCL_ERROR(
874 : "[BuildAsync] transport build async failed, self rank[%u], peer rank[%u]", paraVector_[rank_].worldRank,
875 : paraVector_[dstIntraServerVec_[intraIndex]].worldRank),
876 : HCCL_E_INTERNAL);
877 : }
878 :
879 : // 当前rank作为server端角色
880 0 : u32 dstIntraClientNum = dstIntraClientVec_.size();
881 0 : for (u32 intraIndex = 0; intraIndex < dstIntraClientNum; intraIndex++) {
882 0 : HcclResult ret = TransportBuildAsync(
883 0 : MachineType::MACHINE_SERVER_TYPE, paraVector_[rank_].serverId, dstIntraClientVec_[intraIndex], sockets,
884 0 : transportStatus_[dstIntraClientVec_[intraIndex]]);
885 0 : CHK_PRT_RET(
886 : ret,
887 : HCCL_ERROR(
888 : "[BuildAsync] transport build async failed, self rank[%u], peer rank[%u]", paraVector_[rank_].worldRank,
889 : paraVector_[dstIntraClientVec_[intraIndex]].worldRank),
890 : HCCL_E_INTERNAL);
891 : }
892 :
893 : // 暂不支持 跨node通信
894 :
895 0 : CHK_RET(GetBuildStatus(status));
896 0 : return HCCL_SUCCESS;
897 0 : }
898 :
899 0 : HcclResult CommBase::BuildQuerry(u32& status)
900 : {
901 0 : for (u32 i = 0; i < transportStatus_.size(); i++) {
902 0 : if (transportStatus_[i] == HETEROG_P2P_WAIT) {
903 0 : CHK_RET(TransportBuildQuerry(i, transportStatus_[i]));
904 : }
905 : }
906 0 : CHK_RET(GetBuildStatus(status));
907 0 : HCCL_DEBUG("BuildQuerry: %u", status);
908 0 : return HCCL_SUCCESS;
909 : }
910 :
911 0 : HcclResult CommBase::GetBuildStatus(u32& status)
912 : {
913 0 : u32 transportDoneNum = 0;
914 0 : u32 transportErrorNum = 0;
915 0 : for (u32 i = 0; i < transportStatus_.size(); i++) {
916 0 : if (transportStatus_[i] == HETEROG_P2P_SUCCESS) {
917 0 : transportDoneNum++;
918 0 : } else if (transportStatus_[i] == HETEROG_P2P_FAILED) {
919 0 : transportErrorNum++;
920 : }
921 : }
922 : u32 transportNum
923 0 : = dstInterClientMap_.size() + dstIntraClientVec_.size() + dstInterServerMap_.size() + dstIntraServerVec_.size();
924 0 : if (transportErrorNum > 0) {
925 0 : status = HETEROG_P2P_FAILED;
926 0 : HCCL_ERROR("transport error num[%u].", transportErrorNum);
927 0 : return HCCL_E_INTERNAL;
928 0 : } else if (transportDoneNum == transportNum) {
929 0 : status = HETEROG_P2P_SUCCESS;
930 0 : HCCL_INFO("CommBase connect complete.");
931 0 : } else if (transportDoneNum < transportNum) {
932 0 : status = HETEROG_P2P_WAIT;
933 : } else {
934 0 : status = HETEROG_P2P_FAILED;
935 0 : HCCL_ERROR("transport done num[%u] invalid, expect[%u].", transportDoneNum, transportNum);
936 0 : return HCCL_E_INTERNAL;
937 : }
938 0 : return HCCL_SUCCESS;
939 : }
940 :
941 0 : HcclResult CommBase::TransportBuildAsync(
942 : const MachineType machineType, const std::string& serverId, u32 dstRank,
943 : const std::vector<std::shared_ptr<HcclSocket>>& sockets, u32& status)
944 : {
945 0 : CHK_PRT_RET(
946 : dstRank >= transportInfo_.size(),
947 : HCCL_ERROR("[TransportQuerry] Transport[%u] is invalid, should init before query it.", dstRank), HCCL_E_PARA);
948 0 : MachinePara machinePara;
949 0 : CHK_RET(SetMachinePara(machineType, serverId, dstRank, sockets, machinePara));
950 : // 实例化TransportBase
951 0 : CHK_RET(SetTransportType(dstRank));
952 0 : std::chrono::milliseconds kdefaultTimeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
953 :
954 0 : TransportPara para{};
955 0 : para.timeout = kdefaultTimeout;
956 0 : para.virtualFlag = false;
957 0 : transportInfo_[dstRank].reset(
958 0 : new (std::nothrow) Transport(TransportType::TRANS_TYPE_RESERVED, para, dispatcher_, notifyPool_, machinePara));
959 0 : CHK_PRT_RET(!transportInfo_[dstRank], HCCL_ERROR("[Init][Transport]In create link, new link failed"), HCCL_E_PTR);
960 :
961 0 : CHK_RET(transportInfo_[dstRank]->ConnectAsync(status));
962 0 : if (status == HETEROG_P2P_SUCCESS && checkStatus_[dstRank] == false) {
963 0 : checkStatus_[dstRank] = true;
964 0 : CHK_RET(CheckExchangeInfo(transportInfo_[dstRank], machinePara.localDeviceId));
965 : }
966 0 : HCCL_DEBUG("TransportBuildAsync[%u] %u", dstRank, status);
967 0 : return HCCL_SUCCESS;
968 0 : }
969 :
970 0 : HcclResult CommBase::TransportBuildQuerry(u32 dstRank, u32& status)
971 : {
972 0 : CHK_PRT_RET(
973 : dstRank >= transportInfo_.size(),
974 : HCCL_ERROR("[TransportQuerry] Transport[%u] is invalid, should init before query it.", dstRank), HCCL_E_PARA);
975 0 : if (transportInfo_[dstRank]) {
976 0 : CHK_RET(transportInfo_[dstRank]->ConnectQuerry(status));
977 0 : if (status == HETEROG_P2P_SUCCESS && checkStatus_[dstRank] == false) {
978 0 : checkStatus_[dstRank] = true;
979 0 : CHK_RET(CheckExchangeInfo(transportInfo_[dstRank], paraVector_[rank_].devicePhyId));
980 : }
981 : } else {
982 0 : status = HETEROG_P2P_WAIT;
983 : }
984 0 : HCCL_DEBUG("TransportBuildQuerry[%u] %u", dstRank, status);
985 0 : return HCCL_SUCCESS;
986 : }
987 :
988 0 : HcclResult CommBase::CreateExchangerNetwork()
989 : {
990 0 : CHK_PRT_RET(
991 : dstIntraServerVec_.empty() && dstIntraClientVec_.empty(),
992 : HCCL_DEBUG("[Create][ExchangerNetwork]dstIntraServerVec and dstIntraClientVec is empty, do nothing."),
993 : HCCL_SUCCESS);
994 :
995 0 : bool isInterServer = false; // 是否跨server
996 0 : bool isInterHccs = true; // 是否超节点模式
997 0 : std::map<u32, HcclSocketRole> rankRole;
998 0 : CHK_RET(GetRankLinkInfo(isInterServer, isInterHccs, rankRole));
999 :
1000 : // 保持原逻辑不变，将rank&deviceIP信息，构造成 std::map<u32, std::vector<HcclIpAddress> >，使用
1001 : // SocketManager创建链接
1002 0 : std::string commTag = (Is310PDevice() || isHaveCpuRank_) ? tag_ : collectiveId_;
1003 0 : HcclIpAddress localIP;
1004 0 : std::map<u32, HcclRankLinkInfo> dstServerMap;
1005 0 : std::map<u32, HcclRankLinkInfo> dstClientMap;
1006 0 : bool isSupportReuse = false;
1007 :
1008 0 : CHK_RET(GetRankIPInfo(
1009 : isInterServer, isInterHccs, isSupportReuse, rankRole, localIP, dstServerMap, dstClientMap, socketManager_));
1010 :
1011 0 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> serverSocketsMap;
1012 0 : std::map<u32, std::vector<std::shared_ptr<HcclSocket>>> clientSocketsMap;
1013 :
1014 0 : HcclUs startut = TIME_NOW();
1015 0 : HcclResult ret = socketManager_->CreateSockets(
1016 0 : commTag, false, netDevCtxMap_[localIP], dstServerMap, dstClientMap, serverSocketsMap, clientSocketsMap,
1017 : isSupportReuse);
1018 0 : CHK_PRT_RET(
1019 : ret != HCCL_SUCCESS, HCCL_ERROR("[Create][ExchangerNetwork]sync create connections Failed, ret[%u].", ret),
1020 : ret);
1021 :
1022 0 : HCCL_DEBUG(
1023 : "[Create][Exchanger] serverSocketsMap size[%u], clientSocketsMap size[%u]", serverSocketsMap.size(),
1024 : clientSocketsMap.size());
1025 0 : intraSocketsMap_.insert(serverSocketsMap.begin(), serverSocketsMap.end());
1026 0 : intraSocketsMap_.insert(clientSocketsMap.begin(), clientSocketsMap.end());
1027 :
1028 0 : HCCL_INFO("[Create][ExchangerNetwork]create connections duration time:%lld us.", DURATION_US(TIME_NOW() - startut));
1029 0 : return HCCL_SUCCESS;
1030 0 : }
1031 :
1032 0 : HcclResult CommBase::GetRankIPInfo(
1033 : bool isInterServer, bool isInterHccs, bool& isSupportReuse, std::map<u32, HcclSocketRole>& rankRole,
1034 : HcclIpAddress& localIP, std::map<u32, HcclRankLinkInfo>& dstServerMap,
1035 : std::map<u32, HcclRankLinkInfo>& dstClientMap, std::shared_ptr<HcclSocketManager>& socketManager)
1036 : {
1037 0 : if (Is310PDevice() || isHaveCpuRank_) {
1038 : // 310P和异构场景
1039 0 : std::vector<u32> dstIntraVec;
1040 0 : for (auto it = rankRole.begin(); it != rankRole.end(); ++it) {
1041 0 : dstIntraVec.push_back(it->first);
1042 : }
1043 0 : CHK_RET(GetIntraRankIPInfo(dstIntraVec, localIP, dstServerMap, dstClientMap));
1044 :
1045 : // 不复用，每次都创建
1046 0 : isSupportReuse = false;
1047 0 : socketManager.reset(new (std::nothrow)
1048 0 : HcclSocketManager(nicDeployInner_, deviceLogicId_, devicePhyId_, userRank_));
1049 0 : CHK_PTR_NULL(socketManager);
1050 0 : } else if (isInterServer && isInterHccs) {
1051 : // 超节点间Hccs模式
1052 0 : CHK_RET(GetSuperNodeIntraRankIPInfo(rankRole, localIP, dstServerMap, dstClientMap));
1053 0 : isSupportReuse = true;
1054 0 : socketManager = exchanger_.socketManager;
1055 0 : CHK_PTR_NULL(socketManager);
1056 0 : } else if (isInterServer == false) {
1057 : // server内模式
1058 0 : CHK_RET(GetIntraRankIPInfo(rankRole, localIP, dstServerMap, dstClientMap));
1059 0 : isSupportReuse = true;
1060 0 : socketManager = exchanger_.socketManager;
1061 0 : CHK_PTR_NULL(socketManager);
1062 : } else {
1063 0 : HCCL_ERROR(
1064 : "[Create][ExchangerNetwork]isInterServer[%d] and isInterHccs[%d] is not support", isInterServer,
1065 : isInterHccs);
1066 0 : return HCCL_E_INTERNAL;
1067 : }
1068 0 : return HCCL_SUCCESS;
1069 : }
1070 :
1071 0 : HcclResult CommBase::GetRankLinkInfo(bool& isInterServer, bool& isInterHccs, std::map<u32, HcclSocketRole>& rankRole)
1072 : {
1073 0 : std::vector<u32> devicePhyIds;
1074 0 : for (u32 dstRank : dstIntraServerVec_) {
1075 0 : isInterServer |= (paraVector_[dstRank].serverId != paraVector_[rank_].serverId);
1076 0 : isInterHccs &= IsSupportInterHccs(dstRank);
1077 0 : rankRole.insert(std::make_pair(dstRank, HcclSocketRole::SOCKET_ROLE_SERVER));
1078 0 : devicePhyIds.push_back(paraVector_[dstRank].devicePhyId);
1079 : }
1080 0 : for (u32 dstRank : dstIntraClientVec_) {
1081 0 : isInterServer |= (paraVector_[dstRank].serverId != paraVector_[rank_].serverId);
1082 0 : isInterHccs &= IsSupportInterHccs(dstRank);
1083 0 : rankRole.insert(std::make_pair(dstRank, HcclSocketRole::SOCKET_ROLE_CLIENT));
1084 0 : devicePhyIds.push_back(paraVector_[dstRank].devicePhyId);
1085 : }
1086 0 : rankRole.insert(std::make_pair(rank_, HcclSocketRole::SOCKET_ROLE_RESERVED));
1087 0 : devicePhyIds.push_back(paraVector_[rank_].devicePhyId);
1088 :
1089 0 : if (paraVector_[rank_].deviceType == DevType::DEV_TYPE_310P3) {
1090 0 : HcclResult ret = P2PMgmtPub::EnableP2P(devicePhyIds);
1091 0 : CHK_PRT_RET(
1092 : ret != HCCL_SUCCESS,
1093 : HCCL_ERROR(
1094 : "[Get][RankLinkInfo]Enable P2P Failed, devicePhyId[%d], ret[%u]", paraVector_[rank_].devicePhyId, ret),
1095 : ret);
1096 : }
1097 : // server内非异构场景，使能P2P
1098 : // 心跳需要单独WaitP2PEnabled？
1099 0 : if (!isInterServer && !isHaveCpuRank_) {
1100 0 : HcclResult ret = P2PMgmtPub::WaitP2PEnabled(devicePhyIds);
1101 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[Get][RankLinkInfo]Enable P2P Failed, ret[%u]", ret), ret);
1102 : }
1103 0 : return HCCL_SUCCESS;
1104 0 : }
1105 :
1106 0 : HcclResult CommBase::GetIntraRankIPInfo(
1107 : std::map<u32, HcclSocketRole>& rankRole, HcclIpAddress& localIP, std::map<u32, HcclRankLinkInfo>& dstServerMap,
1108 : std::map<u32, HcclRankLinkInfo>& dstClientMap)
1109 : {
1110 0 : u32 userRankSize = rankRole.size();
1111 0 : for (auto rankIter = rankRole.begin(); rankIter != rankRole.end(); rankIter++) {
1112 0 : u32 dstRank = rankIter->first;
1113 0 : s32 dstDeviceId = useSuperPodMode_ ? static_cast<s32>(paraVector_[dstRank].superDeviceId) :
1114 0 : paraVector_[dstRank].devicePhyId;
1115 0 : HcclSocketRole localRole = rankIter->second;
1116 :
1117 : // Rank devicePhyId 作为地址
1118 0 : HcclRankLinkInfo linkInfo{};
1119 0 : linkInfo.userRank = paraVector_[dstRank].userRank;
1120 0 : linkInfo.devicePhyId = dstDeviceId;
1121 0 : if (vnicRanksPort_.empty() || (userRankSize > 1 && IsSupportMC2(tag_) >= MC2_PLANE_MODE_COMBINE)) {
1122 0 : linkInfo.port = GetNicPort(paraVector_[dstRank].devicePhyId, ranksPort_, linkInfo.userRank, isUseRankPort_);
1123 : } else {
1124 : linkInfo.port
1125 0 : = GetNicPort(paraVector_[dstRank].devicePhyId, vnicRanksPort_, linkInfo.userRank, isUseRankPort_);
1126 : }
1127 0 : HcclIpAddress ipAddress(linkInfo.devicePhyId);
1128 0 : DeviceIdType deviceidType
1129 0 : = useSuperPodMode_ ? (DeviceIdType::DEVICE_ID_TYPE_SDID) : (DeviceIdType::DEVICE_ID_TYPE_PHY_ID);
1130 : // rank个数小于等于1时，没有初始化ra资源，无法调用device侧hccp接口
1131 0 : if (userRankSize > 1) {
1132 0 : if (IsSupportMC2(tag_) >= MC2_PLANE_MODE_COMBINE) {
1133 0 : ipAddress = paraVector_[dstRank].nicIp.front();
1134 0 : CHK_PRT_RET(
1135 : ipAddress.IsInvalid(),
1136 : HCCL_ERROR(
1137 : "[Get][IntraRankIPInfo] ipAddress is invalid when NIC, check the ip configuration for "
1138 : "dstRank[%u]",
1139 : dstRank),
1140 : HCCL_E_PARA);
1141 : } else {
1142 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
1143 : paraVector_[rank_].devicePhyId, deviceidType, linkInfo.devicePhyId, ipAddress));
1144 : }
1145 : }
1146 0 : linkInfo.ip = ipAddress;
1147 0 : linkInfo.socketsPerLink = 1;
1148 :
1149 0 : HCCL_DEBUG(
1150 : "[Get][IntraRankIPInfo] tag[%s], userRank[%u], destRank[%u], localRole[%d], port[%u], ip[%s], "
1151 : "devicePhyId[%u]",
1152 : tag_.c_str(), rank_, linkInfo.userRank, localRole, linkInfo.port, linkInfo.ip.GetReadableAddress(),
1153 : paraVector_[rank_].devicePhyId);
1154 :
1155 0 : if (localRole == HcclSocketRole::SOCKET_ROLE_CLIENT) {
1156 0 : dstServerMap.insert(std::make_pair(linkInfo.userRank, linkInfo));
1157 0 : } else if (localRole == HcclSocketRole::SOCKET_ROLE_SERVER) {
1158 0 : dstClientMap.insert(std::make_pair(linkInfo.userRank, linkInfo));
1159 : } else {
1160 : // 当前上层逻辑，保证 userRank_(当前 Rank) 在 userRanks 中
1161 0 : localIP = linkInfo.ip;
1162 : }
1163 0 : }
1164 0 : return HCCL_SUCCESS;
1165 : }
1166 :
1167 1 : HcclResult CommBase::GetIntraRankIPInfo(
1168 : std::vector<u32>& dstIntraVec, HcclIpAddress& localIP, std::map<u32, HcclRankLinkInfo>& dstServerMap,
1169 : std::map<u32, HcclRankLinkInfo>& dstClientMap)
1170 : {
1171 2 : for (u32 dstRank : dstIntraVec) {
1172 1 : auto& rankInfo = paraVector_[dstRank];
1173 1 : HcclRankLinkInfo linkInfo{};
1174 1 : linkInfo.userRank = rankInfo.userRank;
1175 1 : linkInfo.devicePhyId = rankInfo.devicePhyId;
1176 1 : linkInfo.ip = isHaveCpuRank_ ? rankInfo.hostIp : rankInfo.nicIp[0];
1177 1 : if (!vnicRanksPort_.empty()) {
1178 0 : linkInfo.port = GetNicPort(linkInfo.devicePhyId, vnicRanksPort_, linkInfo.userRank, isUseRankPort_);
1179 : } else {
1180 1 : linkInfo.port = GetNicPort(linkInfo.devicePhyId, ranksPort_, linkInfo.userRank, isUseRankPort_);
1181 : }
1182 1 : linkInfo.socketsPerLink = 1;
1183 :
1184 : HcclSocketRole localRole;
1185 1 : if (paraVector_[rank_].userRank < linkInfo.userRank) {
1186 0 : dstClientMap.insert(std::make_pair(linkInfo.userRank, linkInfo));
1187 0 : localRole = HcclSocketRole::SOCKET_ROLE_CLIENT;
1188 1 : } else if (paraVector_[rank_].userRank > linkInfo.userRank) {
1189 0 : dstServerMap.insert(std::make_pair(linkInfo.userRank, linkInfo));
1190 0 : localRole = HcclSocketRole::SOCKET_ROLE_SERVER;
1191 : } else {
1192 1 : localIP = linkInfo.ip;
1193 1 : localRole = HcclSocketRole::SOCKET_ROLE_RESERVED;
1194 : }
1195 1 : HCCL_DEBUG(
1196 : "[Get][IntraRankIPInfo] userRank[%u], destRank[%u], localRole[%d], port[%u], ip[%s]", userRank_,
1197 : linkInfo.userRank, localRole, linkInfo.port, linkInfo.ip.GetReadableAddress());
1198 1 : }
1199 1 : return HCCL_SUCCESS;
1200 : }
1201 :
1202 0 : HcclResult CommBase::GetSuperNodeIntraRankIPInfo(
1203 : std::map<u32, HcclSocketRole>& rankRole, HcclIpAddress& localIP, std::map<u32, HcclRankLinkInfo>& dstServerMap,
1204 : std::map<u32, HcclRankLinkInfo>& dstClientMap)
1205 : {
1206 0 : for (auto rankIter = rankRole.begin(); rankIter != rankRole.end(); rankIter++) {
1207 0 : u32 dstUserRank = paraVector_[rankIter->first].userRank;
1208 0 : HcclIpAddress ipAddr(paraVector_[rankIter->first].nicIp[0]);
1209 0 : if (!GetExternalInputInterHccsDisable()) {
1210 0 : CHK_RET(hrtRaGetSingleSocketVnicIpInfo(
1211 : devicePhyId_, DeviceIdType::DEVICE_ID_TYPE_SDID, paraVector_[rankIter->first].superDeviceId, ipAddr));
1212 : }
1213 0 : HcclSocketRole localRole = rankIter->second;
1214 :
1215 0 : HcclRankLinkInfo linkInfo{};
1216 0 : u32 dstRank = INVALID_VALUE_RANKID;
1217 0 : linkInfo.userRank = dstUserRank;
1218 0 : linkInfo.devicePhyId = -1;
1219 0 : linkInfo.ip = ipAddr;
1220 0 : CHK_RET(GetRankByUserRank(linkInfo.userRank, dstRank));
1221 0 : if (!vnicRanksPort_.empty()) {
1222 : linkInfo.port
1223 0 : = GetNicPort(paraVector_[dstRank].devicePhyId, vnicRanksPort_, linkInfo.userRank, isUseRankPort_);
1224 : } else {
1225 0 : linkInfo.port = GetNicPort(paraVector_[dstRank].devicePhyId, ranksPort_, linkInfo.userRank, isUseRankPort_);
1226 : }
1227 0 : linkInfo.socketsPerLink = 1;
1228 :
1229 0 : HCCL_DEBUG(
1230 : "[Get][SuperNodeIntraRankIPInfo] userRank[%u], destRank[%u], localRole[%d], port[%u], ip[%s]", userRank_,
1231 : dstUserRank, localRole, linkInfo.port, linkInfo.ip.GetReadableAddress());
1232 :
1233 0 : if (localRole == HcclSocketRole::SOCKET_ROLE_CLIENT) {
1234 0 : dstServerMap.insert(std::make_pair(dstUserRank, linkInfo));
1235 0 : } else if (localRole == HcclSocketRole::SOCKET_ROLE_SERVER) {
1236 0 : dstClientMap.insert(std::make_pair(dstUserRank, linkInfo));
1237 : } else {
1238 : // 当前上层逻辑，保证 userRank_(当前 Rank) 在 userRanks 中
1239 0 : localIP = linkInfo.ip;
1240 : }
1241 0 : }
1242 0 : return HCCL_SUCCESS;
1243 : }
1244 :
1245 1 : bool CommBase::IsSupportInterHccs(const u32 dstRank)
1246 : {
1247 : // 仅判断超节点内, 兼容打平通信域同时有server内和server间, 因此不判断server_id
1248 1 : bool isInterHccs = GetExternalInputInterHccsDisable() == false
1249 1 : && paraVector_[rank_].deviceType == DevType::DEV_TYPE_910_93
1250 0 : && paraVector_[rank_].superPodId.empty() == false
1251 2 : && paraVector_[rank_].superPodId == paraVector_[dstRank].superPodId;
1252 :
1253 1 : HCCL_INFO(
1254 : "[IsSupportInterHccs]rank[%u], superPodId[%s], dstRank[%u], dstSuperPodId[%s], isInterHccs[%d]", rank_,
1255 : paraVector_[rank_].superPodId.c_str(), dstRank, paraVector_[dstRank].superPodId.c_str(), isInterHccs);
1256 1 : return isInterHccs;
1257 : }
1258 :
1259 2 : void CommBase::SetMachineLinkMode(MachinePara& machinePara) { machinePara.linkMode = LinkMode::LINK_DUPLEX_MODE; }
1260 :
1261 10 : HcclResult CommBase::SetHDCModeInfo(
1262 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>>& rankDevicePhyIdNicInfoMap,
1263 : std::vector<u32>& ranksPort, std::vector<u32>& vnicRanksPort, bool isSetHDCModeInfo, bool isUseRankPort)
1264 : {
1265 10 : rankDevicePhyIdNicInfoMap_ = rankDevicePhyIdNicInfoMap;
1266 10 : ranksPort_ = ranksPort;
1267 10 : vnicRanksPort_ = vnicRanksPort;
1268 10 : isSetHDCModeInfo_ = isSetHDCModeInfo;
1269 10 : isUseRankPort_ = isUseRankPort;
1270 10 : return HCCL_SUCCESS;
1271 : }
1272 :
1273 0 : HcclResult CommBase::CheckExchangeInfo(const std::shared_ptr<Transport>& link, const s32 deviceId)
1274 : {
1275 : // 算子一致性校验
1276 0 : u64 exchangeInfoLength = RankConsistentcyChecker::GetInstance(deviceId).GetRankConsistentDataLength();
1277 0 : std::vector<u8> recvData = link->GetExchangeInfo();
1278 0 : if (recvData.size() != 0) {
1279 0 : CHK_PRT_RET(
1280 : recvData.size() != exchangeInfoLength,
1281 : HCCL_ERROR(
1282 : "[Check][ExchangeInfo]remote exchangInfo size[%zu], local exchangeInfo size[%llu]", recvData.size(),
1283 : exchangeInfoLength),
1284 : HCCL_E_INTERNAL);
1285 0 : CHK_RET(
1286 : RankConsistentcyChecker::GetInstance(deviceId).CheckFrameRecv(&recvData[0], recvData.size(), tag_.c_str()));
1287 : }
1288 :
1289 0 : return HCCL_SUCCESS;
1290 0 : }
1291 :
1292 0 : u32 CommBase::IsSupportMC2(const std::string& tag)
1293 : {
1294 0 : const std::string& suffix = HCCL_MC2_MULTISERVER_SUFFIX;
1295 0 : u32 mc2MultiServerType = MC2_PLANE_MODE_HOST; // 非直驱场景
1296 0 : if (tag.size() > suffix.size() && tag.compare(tag.size() - suffix.size(), suffix.size(), suffix) == 0) {
1297 0 : mc2MultiServerType = MC2_PLANE_MODE_COMBINE; // 非分层建链场景
1298 0 : if (tag.find("_HIE") != std::string::npos) {
1299 0 : mc2MultiServerType = MC2_PLANE_MODE_HIERARCHY; // 分层建链场景
1300 : }
1301 : }
1302 0 : return mc2MultiServerType;
1303 : }
1304 : } // namespace hccl
|
