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_factory.h"
12 : #include <sstream>
13 : #include <algorithm>
14 : namespace hccl {
15 :
16 531 : CommFactory::CommFactory(const std::string &identifier, const u32 userRank, const u32 userRankSize,
17 : const HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool> ¬ifyPool,
18 : std::map<HcclIpAddress, HcclNetDevCtx> &netDevCtxMap,
19 : std::shared_ptr<TopoInfoExtractor> topoInfoEx,
20 : const bool isUsedRdmaLevel0, const TopoType topoFlag, const DevType deviceType,
21 : const std::vector<RankInfo> rankVector, const NICDeployment nicDeploymentInner, bool isHeterogComm,
22 531 : u32 meshAggregationRankSize, bool isHaveCpuRank, bool isUsedInterHccsMode, bool useSuperPodMode)
23 531 : : identifier_(identifier),
24 531 : userRank_(userRank),
25 531 : userRankSize_(userRankSize),
26 531 : topoFlag_(topoFlag),
27 531 : deviceType_(deviceType),
28 531 : dispatcher_(dispatcher),
29 531 : notifyPool_(notifyPool),
30 531 : netDevCtxMap_(netDevCtxMap),
31 531 : topoInfoEx_(topoInfoEx),
32 531 : isUsedRdmaLevel0_(isUsedRdmaLevel0),
33 531 : rankVector_(rankVector),
34 531 : nicDeployInner_(nicDeploymentInner),
35 531 : isHeterogComm_(isHeterogComm),
36 531 : isHaveCpuRank_(isHaveCpuRank),
37 531 : reusedSocketManager_(),
38 531 : deviceLogicId_(0),
39 531 : isUsedInterHccsMode_(isUsedInterHccsMode),
40 1593 : useSuperPodMode_(useSuperPodMode)
41 : {
42 : (void) meshAggregationRankSize;
43 531 : }
44 :
45 1563 : CommFactory::~CommFactory()
46 : {
47 : // 销毁资源
48 520 : CommPlaneVector_.clear();
49 525 : isBridgeVector_.clear();
50 526 : superPodToRank_.clear();
51 530 : serverToRank_.clear();
52 531 : deviceLinkTypeMap_.clear();
53 529 : rankVector_.clear();
54 1043 : }
55 :
56 530 : HcclResult CommFactory::Init()
57 : {
58 530 : CHK_RET(topoInfoEx_->CheckInitInfo());
59 :
60 527 : topoInfoEx_->GetCommPlaneVector(CommPlaneVector_);
61 527 : topoInfoEx_->GetIsBridgeVector(isBridgeVector_);
62 527 : topoInfoEx_->GetRankData(rankData_);
63 527 : topoInfoEx_->GetServerToRank(serverToRank_);
64 527 : topoInfoEx_->GetSuperPodToRank(superPodToRank_);
65 527 : topoInfoEx_->GetDeviceLinkTypeMap(deviceLinkTypeMap_);
66 :
67 527 : s32 deviceLogicID = 0;
68 527 : if (!isHeterogComm_ && rankVector_[userRank_].devicePhyId != HOST_DEVICE_ID) {
69 527 : CHK_RET(hrtGetDevice(&deviceLogicID));
70 527 : deviceLogicId_ = deviceLogicID;
71 : }
72 :
73 527 : reusedSocketManager_.reset(new (std::nothrow) HcclSocketManager(nicDeployInner_, deviceLogicId_,
74 527 : rankVector_[userRank_].devicePhyId, userRank_));
75 527 : CHK_PTR_NULL(reusedSocketManager_);
76 :
77 527 : return HCCL_SUCCESS;
78 : }
79 :
80 30 : HcclResult CommFactory::CheckCommPara(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
81 : const CommParaInfo &commParaInfo)
82 : {
83 : (void) inputMem;
84 : (void) outputMem;
85 30 : CHK_PRT_RET(commParaInfo.commPlane >= COMM_LEVEL_RESERVED,
86 : HCCL_ERROR("[Check][CommPara]tag[%s], commPlane[%d] is invalid, is out of range [0, %d]",
87 : tag.c_str(), commParaInfo.commPlane, COMM_LEVEL_RESERVED - 1), HCCL_E_PARA);
88 :
89 : // 判断commPlane和commType的组合是否支持
90 30 : bool isSupport = true;
91 30 : switch (commParaInfo.commType) {
92 12 : case CommType::COMM_TAG_RING_INNER:
93 : case CommType::COMM_TAG_HALVING_DOUBLING: {
94 35 : isSupport = (commParaInfo.commPlane == COMM_LEVEL0) ||
95 18 : (commParaInfo.commPlane == COMM_LEVEL1) ||
96 6 : (commParaInfo.commPlane == COMM_LEVEL2);
97 12 : break;
98 : }
99 1 : case CommType::COMM_TAG_MESH: {
100 3 : isSupport = (commParaInfo.commPlane == COMM_COMBINE) ||
101 1 : (commParaInfo.commPlane == COMM_LEVEL0) ||
102 0 : (commParaInfo.commPlane == COMM_MESH_L0) ||
103 0 : (commParaInfo.commPlane == COMM_MESH_L1) ||
104 2 : (commParaInfo.commPlane == COMM_LEVEL2) ||
105 0 : (commParaInfo.commPlane == COMM_COMBINE_ORDER);
106 1 : break;
107 : }
108 1 : case CommType::COMM_TAG_RING_COMBINED:
109 : case CommType::COMM_TAG_P2P: {
110 1 : isSupport = commParaInfo.commPlane == COMM_COMBINE;
111 1 : break;
112 : }
113 1 : case CommType::COMM_TAG_MESH_COMBINED: {
114 1 : isSupport = commParaInfo.commPlane == COMM_COMBINE_ORDER;
115 1 : break;
116 : }
117 4 : case CommType::COMM_TAG_ASYMMETRIC_HIERARCHICAL_CONCATENATE:
118 : case CommType::COMM_TAG_ASYMMETRIC_HIERARCHICAL_CONCATENATE_BROKE:
119 : case CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING:
120 : case CommType::COMM_TAG_NONUNIFORM_BRUCK: {
121 4 : isSupport = (commParaInfo.commPlane == COMM_LEVEL1);
122 4 : break;
123 : }
124 1 : case CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING_V1: {
125 1 : isSupport = (commParaInfo.commPlane == COMM_LEVEL1 && deviceType_ != DevType::DEV_TYPE_910_93);
126 1 : break;
127 : }
128 0 : case CommType::COMM_TAG_STAR:
129 0 : break;
130 9 : case CommType::COMM_TAG_WHOLE_NHR:
131 : case CommType::COMM_TAG_WHOLE_NHR_V1:
132 : case CommType::COMM_TAG_WHOLE_AHC:
133 : case CommType::COMM_TAG_WHOLE_AHC_BROKE:
134 : case CommType::COMM_TAG_WHOLE_NB: {
135 9 : isSupport = (deviceType_ != DevType::DEV_TYPE_910_93);
136 9 : break;
137 : }
138 1 : default: {
139 1 : HCCL_ERROR("[Check][CommPara]commType[%d] is invalid", commParaInfo.commType);
140 1 : return HCCL_E_PARA;
141 : }
142 : }
143 :
144 29 : CHK_PRT_RET(isSupport == false,
145 : HCCL_ERROR("[Check][CommPara]tag[%s], deviceType[%d], commPlane[%d] and commType[%d] is not support",
146 : tag.c_str(), deviceType_, commParaInfo.commPlane, commParaInfo.commType), HCCL_E_PARA);
147 :
148 29 : return HCCL_SUCCESS;
149 : }
150 :
151 30 : HcclResult CommFactory::CreateCommPlane(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
152 : const CommParaInfo &commParaInfo, std::vector<std::unique_ptr<CommBase> > &commVec, DeviceMem expMem)
153 : {
154 30 : HcclUs startut = TIME_NOW();
155 30 : HcclResult ret = HCCL_SUCCESS;
156 30 : HCCL_INFO("[Create][CommPlane]tag[%s], identifier[%s], commPlane[%d], commType[%d]",
157 : tag.c_str(), identifier_.c_str(), commParaInfo.commPlane, commParaInfo.commType);
158 :
159 30 : CHK_RET(CheckCommPara(tag, inputMem, outputMem, commParaInfo));
160 29 : bool isUsedRdma = false;
161 29 : CHK_RET(GetIsUsedRdma(commParaInfo, isUsedRdma));
162 :
163 29 : switch (commParaInfo.commType) {
164 17 : case CommType::COMM_TAG_RING_INNER:
165 : case CommType::COMM_TAG_RING_COMBINED:
166 : case CommType::COMM_TAG_NONUNIFORM_BRUCK:
167 : case CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING:
168 : case CommType::COMM_TAG_NONUNIFORM_HIERARCHICAL_RING_V1:{
169 17 : ret = CreateCommRing(tag, inputMem, outputMem, commParaInfo, CommPlaneVector_[commParaInfo.commPlane],
170 : isUsedRdma, commVec);
171 17 : break;
172 : }
173 1 : case CommType::COMM_TAG_HALVING_DOUBLING: {
174 1 : ret = CreateCommHD(tag, inputMem, outputMem, commParaInfo, CommPlaneVector_[commParaInfo.commPlane],
175 : isUsedRdma, commVec);
176 1 : break;
177 : }
178 0 : case CommType::COMM_TAG_STAR: {
179 0 : std::vector<std::vector<RankInfo> > commPlaneVec;
180 0 : std::vector<RankInfo> linkParas;
181 0 : CreateStarLinkPara(linkParas);
182 0 : commPlaneVec.push_back(linkParas);
183 0 : ret = CreateCommStar(tag, inputMem, outputMem, commParaInfo, commPlaneVec, isUsedRdma, commVec);
184 0 : break;
185 0 : }
186 2 : case CommType::COMM_TAG_MESH:
187 : case CommType::COMM_TAG_MESH_COMBINED: {
188 2 : if (commParaInfo.meshSinglePlane == true) {
189 : // 910B非确定性计算场景，server内MESH组网只需要创建一个commbase平面
190 0 : std::vector<std::vector<RankInfo> > commPlaneVec;
191 0 : commPlaneVec.push_back(CommPlaneVector_[commParaInfo.commPlane][0]);
192 0 : ret = CreateCommMesh(tag, inputMem, outputMem, commParaInfo, commPlaneVec, isUsedRdma, commVec, expMem);
193 0 : } else {
194 2 : ret = CreateCommMesh(tag, inputMem, outputMem, commParaInfo, CommPlaneVector_[commParaInfo.commPlane],
195 : isUsedRdma, commVec, expMem);
196 : }
197 2 : break;
198 : }
199 0 : case CommType::COMM_TAG_P2P: {
200 0 : ret = CreateCommP2P(tag, inputMem, outputMem, commParaInfo,
201 0 : CommPlaneVector_[commParaInfo.commPlane], isUsedRdma, commVec);
202 0 : break;
203 : }
204 9 : default: {
205 9 : HCCL_ERROR("[Create][CommPlane]commType[%d] is invalid", commParaInfo.commType);
206 9 : return HCCL_E_PARA;
207 : }
208 : }
209 :
210 20 : CHK_PRT_RET(ret != HCCL_SUCCESS,
211 : HCCL_ERROR("[Create][CommPlane]failed, tag[%s], commPlane[%d], commType[%d]",
212 : tag.c_str(), commParaInfo.commPlane, commParaInfo.commType), ret);
213 :
214 20 : HCCL_INFO("complete commPlane[%d] commType[%d] creation, Time:%lld us",
215 : commParaInfo.commPlane, commParaInfo.commType, DURATION_US(TIME_NOW() - startut));
216 20 : return HCCL_SUCCESS;
217 : }
218 :
219 29 : HcclResult CommFactory::GetIsUsedRdma(const CommParaInfo &commParaInfo, bool &isUsedRdma)
220 : {
221 29 : std::vector<std::vector<RankInfo> > commP2PPlaneVec;
222 29 : if (commParaInfo.commType == CommType::COMM_TAG_P2P) {
223 : // P2P只需要判断两张卡之间的连接关系
224 0 : bool invalidcheck = (rankVector_.size() <= userRank_) || (rankVector_.size() <= commParaInfo.peerUserRank);
225 0 : CHK_PRT_RET(invalidcheck, HCCL_ERROR("[GetIsUsedRdma]dstUserRank[%u] or userRank[%u] is bigger than "\
226 : "rankVector size[%u]", commParaInfo.peerUserRank, userRank_, rankVector_.size()), HCCL_E_PARA);
227 :
228 0 : std::vector<RankInfo> commP2PRankVec;
229 0 : commP2PRankVec.push_back(rankVector_[userRank_]);
230 0 : commP2PRankVec.push_back(rankVector_[commParaInfo.peerUserRank]);
231 0 : commP2PPlaneVec.push_back(commP2PRankVec);
232 0 : }
233 :
234 29 : std::vector<std::vector<RankInfo> > &commPlaneVec = (commParaInfo.commType == CommType::COMM_TAG_P2P) ?
235 29 : commP2PPlaneVec : CommPlaneVector_[commParaInfo.commPlane];
236 :
237 29 : bool isInterSuperPod = false;
238 29 : bool isInterServer = false;
239 29 : bool isConnectedWithPcie = false;
240 44 : for (const std::vector<RankInfo> &commPlane : commPlaneVec) {
241 46 : for (const RankInfo &dstRank : commPlane) {
242 31 : if (rankData_.superPodId != dstRank.superPodId) { // 跨超节点场景
243 0 : isInterSuperPod = true;
244 31 : } else if (rankData_.serverIdx != dstRank.serverIdx) { // 不跨超节点, 跨server场景
245 4 : isInterServer = true;
246 : } else { // 同server, PCIE互连场景
247 27 : auto it = deviceLinkTypeMap_.find(dstRank.devicePhyId);
248 27 : CHK_PRT_RET(it == deviceLinkTypeMap_.end(),
249 : HCCL_ERROR("can't find devicePhyId[%d] in deviceLinkTypeMap_", dstRank.devicePhyId),
250 : HCCL_E_NOT_FOUND);
251 27 : isConnectedWithPcie |= (it->second == LinkTypeInServer::PXI_TYPE);
252 : }
253 : }
254 : }
255 : // 使能RDMA的场景: 1.跨超节点 2.跨server且不使能HCCS 3.PCIE连接且使能RDMA开关
256 29 : isUsedRdma = (isInterSuperPod) ||
257 58 : (isInterServer && !isUsedInterHccsMode_) || (isConnectedWithPcie && isUsedRdmaLevel0_);
258 29 : HCCL_INFO("[GetIsUsedRdma]isUsedRdma[%d], isInterSuperPod[%d], isInterServer[%d], isUsedInterHccsMode_[%d], "\
259 : "isConnectedWithPcie[%d], isUsedRdmaLevel0_[%d]", isUsedRdma, isInterSuperPod, isInterServer,
260 : isUsedInterHccsMode_, isConnectedWithPcie, isUsedRdmaLevel0_);
261 29 : return HCCL_SUCCESS;
262 29 : }
263 :
264 17 : HcclResult CommFactory::CreateCommRing(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
265 : const CommParaInfo &commParaInfo, const std::vector<std::vector<RankInfo> > &commPlaneVec,
266 : bool isUsedRdma, std::vector<std::unique_ptr<CommBase> > &commVec)
267 : {
268 17 : u32 ringSize = commPlaneVec.size();
269 17 : commVec.resize(ringSize);
270 :
271 26 : for (u32 ringIndex = 0; ringIndex < ringSize; ++ringIndex) {
272 : // 只有在当前环是bridge rank才需要创建comm实例
273 9 : if (commParaInfo.commPlane == COMM_LEVEL1 && !isBridgeVector_[ringIndex]) {
274 0 : continue; // 跳出本次循环
275 : }
276 :
277 9 : u32 rank = GetSubCollectiveRank(commPlaneVec[ringIndex]);
278 9 : HCCL_DEBUG("[CommFactory][CreateCommRing]rank is %u", rank);
279 9 : if (rank == INVALID_VALUE_RANKID) {
280 0 : continue;
281 : }
282 :
283 9 : IntraExchanger exchangerNetwork {};
284 9 : exchangerNetwork.socketManager = reusedSocketManager_;
285 :
286 9 : HCCL_INFO("[Create][CommRing]comm is used %s. userRank = %u, rank = %u",
287 : isUsedRdma ? "rdma" : "sdma", userRank_, rank);
288 :
289 27 : commVec[ringIndex].reset(new (std::nothrow) CommRing(identifier_, userRank_, userRankSize_,
290 9 : rank, commPlaneVec[ringIndex].size(), topoFlag_, dispatcher_, notifyPool_, netDevCtxMap_, exchangerNetwork,
291 9 : commPlaneVec[ringIndex], inputMem, outputMem, isUsedRdma,
292 : tag, nicDeployInner_,
293 36 : false, false, isHaveCpuRank_, useSuperPodMode_));
294 :
295 9 : CHK_PRT_RET(!commVec[ringIndex], HCCL_ERROR("[Create][CommRing]comm array[%u] reset failed",
296 : ringIndex), HCCL_E_PARA);
297 :
298 9 : if (JudgmentSetHeterogP2p(rank)) {
299 0 : commVec[ringIndex]->SetHeterogP2PType();
300 : }
301 9 : commVec[ringIndex]->SetHDCModeInfo(rankDevicePhyIdNicInfoMap_, ranksPort_,
302 9 : vnicRanksPort_, isSetHDCModeInfo_, isUseRankPort_);
303 9 : if (commVec[ringIndex]->Init() != HCCL_SUCCESS) {
304 0 : HCCL_ERROR("[Create][CommRing]comm array[%u] init failed", ringIndex);
305 0 : commVec[ringIndex].reset(nullptr);
306 0 : return HCCL_E_PARA;
307 : }
308 9 : }
309 17 : return HCCL_SUCCESS;
310 : }
311 :
312 1 : HcclResult CommFactory::CreateCommHD(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
313 : const CommParaInfo &commParaInfo, const std::vector<std::vector<RankInfo> > &commPlaneVec,
314 : bool isUsedRdma, std::vector<std::unique_ptr<CommBase> > &commVec)
315 : {
316 1 : u32 ringSize = commPlaneVec.size();
317 1 : commVec.resize(ringSize);
318 :
319 1 : u32 subUserRankRoot = INVALID_VALUE_RANKID;
320 1 : if (commParaInfo.root != INVALID_VALUE_RANKID) {
321 0 : subUserRankRoot = GetSubRootUserRank(userRank_, commParaInfo.root);
322 0 : if (subUserRankRoot == INVALID_VALUE_RANKID) {
323 0 : HCCL_ERROR("[create][CommHD]get sub root userrank value[%u] invalid.", subUserRankRoot);
324 0 : return HCCL_E_PARA;
325 : }
326 : }
327 :
328 1 : for (u32 ringIndex = 0; ringIndex < ringSize; ++ringIndex) {
329 : // 只有在当前环是bridge rank才需要创建comm实例
330 0 : if (commParaInfo.commPlane == COMM_LEVEL1 && !isBridgeVector_[ringIndex]) {
331 0 : continue; // 跳出本次循环
332 : }
333 :
334 0 : u32 rank = GetSubCollectiveRank(commPlaneVec[ringIndex]);
335 0 : if (rank == INVALID_VALUE_RANKID) {
336 0 : continue;
337 : }
338 :
339 0 : IntraExchanger exchangerNetwork {};
340 0 : exchangerNetwork.socketManager = reusedSocketManager_;
341 :
342 0 : HCCL_INFO("[create][CommHD]comm is used %s. userRank = %u, rank = %u",
343 : isUsedRdma ? "rdma" : "sdma", userRank_, rank);
344 :
345 0 : commVec[ringIndex].reset(new (std::nothrow) CommHalvingDoubling(identifier_, userRank_, userRankSize_,
346 0 : rank, commPlaneVec[ringIndex].size(), topoFlag_, dispatcher_, notifyPool_, netDevCtxMap_, exchangerNetwork,
347 0 : commPlaneVec[ringIndex], inputMem, outputMem, isUsedRdma,
348 : tag, nicDeployInner_, subUserRankRoot, HalvingDoublingType::RECURSIVE_HALVING_DOUBLING,
349 0 : isHaveCpuRank_, useSuperPodMode_));
350 :
351 0 : CHK_PRT_RET(!commVec[ringIndex], HCCL_ERROR("[create][CommHD]comm array[%u] reset failed",
352 : ringIndex), HCCL_E_PARA);
353 :
354 0 : if (JudgmentSetHeterogP2p(rank)) {
355 0 : commVec[ringIndex]->SetHeterogP2PType();
356 : }
357 0 : commVec[ringIndex]->SetHDCModeInfo(rankDevicePhyIdNicInfoMap_, ranksPort_,
358 0 : vnicRanksPort_, isSetHDCModeInfo_, isUseRankPort_);
359 0 : if (commVec[ringIndex]->Init() != HCCL_SUCCESS) {
360 0 : HCCL_ERROR("[create][CommHD]comm array[%u] init failed", ringIndex);
361 0 : commVec[ringIndex].reset(nullptr);
362 0 : return HCCL_E_PARA;
363 : }
364 0 : }
365 1 : return HCCL_SUCCESS;
366 : }
367 :
368 0 : void CommFactory::CreateStarLinkPara(std::vector<RankInfo> &linkParas)
369 : {
370 0 : linkParas = rankVector_;
371 0 : }
372 :
373 0 : HcclResult CommFactory::CreateCommStar(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
374 : const CommParaInfo &commParaInfo, const std::vector<std::vector<RankInfo> > &commPlaneVec,
375 : bool isUsedRdma, std::vector<std::unique_ptr<CommBase> > &commVec)
376 : {
377 0 : HCCL_INFO("create comm star start");
378 0 : u32 ringSize = commPlaneVec.size();
379 0 : commVec.resize(ringSize);
380 :
381 0 : for (u32 ringIndex = 0; ringIndex < ringSize; ++ringIndex) {
382 0 : IntraExchanger exchangerNetwork {};
383 0 : HCCL_INFO("[CreateCommStar] CommStar is used %s. userRank = %u", isUsedRdma ? "rdma" : "sdma", userRank_);
384 :
385 0 : commVec[ringIndex].reset(new (std::nothrow) CommStar(identifier_, userRank_, userRankSize_, userRank_,
386 0 : commPlaneVec[ringIndex].size(), topoFlag_, dispatcher_, notifyPool_, netDevCtxMap_, exchangerNetwork,
387 0 : commPlaneVec[ringIndex], inputMem, outputMem, isUsedRdma, tag, nicDeployInner_, commParaInfo.root, isHaveCpuRank_));
388 :
389 0 : CHK_PRT_RET(!commVec[ringIndex], HCCL_ERROR("[create][CommStar]comm array[%u] reset failed",
390 : ringIndex), HCCL_E_PARA);
391 0 : commVec[ringIndex]->SetHDCModeInfo(rankDevicePhyIdNicInfoMap_, ranksPort_,
392 0 : vnicRanksPort_, isSetHDCModeInfo_, isUseRankPort_);
393 :
394 0 : if (JudgmentSetHeterogP2p(userRank_)) {
395 0 : commVec[ringIndex]->SetHeterogP2PType();
396 : }
397 0 : if (commVec[ringIndex]->Init() != HCCL_SUCCESS) {
398 0 : HCCL_ERROR("[create][CommStar]comm array[%u] star rank[%u] init failed", ringIndex, userRank_);
399 0 : commVec[ringIndex].reset(nullptr);
400 0 : return HCCL_E_PARA;
401 : }
402 0 : }
403 0 : return HCCL_SUCCESS;
404 : }
405 :
406 3 : HcclResult CommFactory::CreateCommMesh(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
407 : const CommParaInfo &commParaInfo, const std::vector<std::vector<RankInfo> > &commPlaneVec, bool isUsedRdma,
408 : std::vector<std::unique_ptr<CommBase> > &commVec, DeviceMem &expMem)
409 : {
410 3 : u32 ringSize = commPlaneVec.size();
411 3 : commVec.resize(ringSize);
412 3 : bool intraIsUseRdma = isUsedRdma;
413 3 : bool isAlltoAllCommMesh = false;
414 :
415 4 : for (u32 ringIndex = 0; ringIndex < ringSize; ++ringIndex) {
416 1 : u32 rank = GetSubCollectiveRank(commPlaneVec[ringIndex]);
417 1 : CHK_PRT_RET(rank == INVALID_VALUE_RANKID, HCCL_ERROR("[Create][CommMesh] invalid rank info."), HCCL_E_PARA);
418 :
419 1 : IntraExchanger exchangerNetwork {};
420 1 : exchangerNetwork.socketManager = reusedSocketManager_;
421 :
422 1 : u32 mc2MultiServerType = commVec[ringIndex]->IsSupportMC2(tag);
423 1 : HCCL_INFO("[Create][CommMesh]comm is used %s. userRank = %u, rank = %u, mc2MultiServerType = %u",
424 : isUsedRdma ? "rdma" : "sdma", userRank_, rank, mc2MultiServerType);
425 1 : if (mc2MultiServerType == MC2_PLANE_MODE_HIERARCHY) { // 分层场景
426 1 : intraIsUseRdma = false;
427 1 : isAlltoAllCommMesh = false;
428 0 : } else if (mc2MultiServerType == MC2_PLANE_MODE_COMBINE) { // 非分层场景
429 0 : intraIsUseRdma = true;
430 : }
431 1 : HCCL_INFO("[Create][CommMesh]comm is created by intraIsUseRdma = %u, isAlltoAllCommMesh = %u",
432 : intraIsUseRdma, isAlltoAllCommMesh);
433 :
434 3 : commVec[ringIndex].reset(new (std::nothrow) CommMesh(identifier_, userRank_, userRankSize_,
435 1 : rank, commPlaneVec[ringIndex].size(), topoFlag_, dispatcher_, notifyPool_, netDevCtxMap_, exchangerNetwork,
436 1 : commPlaneVec[ringIndex], inputMem, outputMem, intraIsUseRdma, tag, isAlltoAllCommMesh, nicDeployInner_, false, commParaInfo.isAicpuModeEn,
437 4 : isHaveCpuRank_, useSuperPodMode_, expMem));
438 :
439 1 : CHK_PRT_RET(!commVec[ringIndex], HCCL_ERROR("[Create][CommMesh]comm array[%u] reset failed",
440 : ringIndex), HCCL_E_PARA);
441 :
442 1 : if (JudgmentSetHeterogP2p(rank)) {
443 0 : commVec[ringIndex]->SetHeterogP2PType();
444 : }
445 1 : commVec[ringIndex]->SetHDCModeInfo(rankDevicePhyIdNicInfoMap_, ranksPort_,
446 1 : vnicRanksPort_, isSetHDCModeInfo_, isUseRankPort_);
447 1 : if (commVec[ringIndex]->Init() != HCCL_SUCCESS) {
448 0 : HCCL_ERROR("[Create][CommMesh]comm array[%u] init failed", ringIndex);
449 0 : commVec[ringIndex].reset(nullptr);
450 0 : return HCCL_E_PARA;
451 : }
452 1 : }
453 3 : return HCCL_SUCCESS;
454 : }
455 :
456 0 : HcclResult CommFactory::CreateCommP2P(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
457 : const CommParaInfo &commParaInfo, const std::vector<std::vector<RankInfo> > &commPlaneVec, bool isUsedRdma,
458 : std::vector<std::unique_ptr<CommBase> > &commVec)
459 : {
460 0 : bool invalidcheck = (rankVector_.size() <= userRank_) || (rankVector_.size() <= commParaInfo.peerUserRank);
461 0 : CHK_PRT_RET(invalidcheck,
462 : HCCL_ERROR("[Create][CommP2P]dstUserRank[%u] or userRank[%u] is bigger than rank vector size[%u].",
463 : commParaInfo.peerUserRank, userRank_, rankVector_.size()), HCCL_E_PARA);
464 :
465 0 : bool heterogP2P = ((rankVector_[userRank_].devicePhyId == HOST_DEVICE_ID) &&
466 0 : (rankVector_[commParaInfo.peerUserRank].devicePhyId != HOST_DEVICE_ID)) ||
467 0 : ((rankVector_[userRank_].devicePhyId != HOST_DEVICE_ID) &&
468 0 : (rankVector_[commParaInfo.peerUserRank].devicePhyId == HOST_DEVICE_ID));
469 :
470 0 : if (heterogP2P) {
471 0 : return CreateCommP2PSync(tag, inputMem, outputMem, commParaInfo, CommPlaneVector_[commParaInfo.commPlane],
472 0 : isUsedRdma, commVec);
473 : }
474 :
475 0 : u32 ringSize = commPlaneVec.size();
476 0 : commVec.resize(ringSize);
477 :
478 0 : for (u32 ringIndex = 0; ringIndex < ringSize; ++ringIndex) {
479 0 : u32 rank = GetSubCollectiveRank(commPlaneVec[ringIndex]);
480 0 : CHK_PRT_RET(rank == INVALID_VALUE_RANKID, HCCL_ERROR("[Create][CommP2P] invalid rank info."), HCCL_E_PARA);
481 :
482 0 : IntraExchanger exchangerNetwork {};
483 0 : exchangerNetwork.socketManager = reusedSocketManager_;
484 :
485 0 : HCCL_INFO("[Create][CommP2P]comm is used %s. userRank = %u, rank = %u",
486 : isUsedRdma ? "rdma" : "sdma", userRank_, rank);
487 :
488 0 : commVec[ringIndex].reset(new (std::nothrow) CommP2P(identifier_, userRank_, userRankSize_,
489 0 : rank, commPlaneVec[ringIndex].size(), topoFlag_, dispatcher_, notifyPool_, netDevCtxMap_, exchangerNetwork,
490 0 : commPlaneVec[ringIndex], inputMem, outputMem, isUsedRdma, tag, commParaInfo.peerUserRank, nicDeployInner_,
491 0 : isHaveCpuRank_, useSuperPodMode_));
492 :
493 0 : CHK_PRT_RET(!commVec[ringIndex], HCCL_ERROR("[Create][CommP2P]comm array[%u] reset failed",
494 : ringIndex), HCCL_E_PARA);
495 :
496 0 : if (commVec[ringIndex]->Init() != HCCL_SUCCESS) {
497 0 : HCCL_ERROR("[Create][CommP2P]comm array[%u] init failed", ringIndex);
498 0 : commVec[ringIndex].reset(nullptr);
499 0 : return HCCL_E_PARA;
500 : }
501 0 : }
502 0 : return HCCL_SUCCESS;
503 : }
504 :
505 0 : HcclResult CommFactory::CreateCommP2PSync(const std::string &tag, const DeviceMem &inputMem, const DeviceMem &outputMem,
506 : const CommParaInfo &commParaInfo, const std::vector<std::vector<RankInfo> > &commPlaneVec, bool isUsedRdma,
507 : std::vector<std::unique_ptr<CommBase> > &commVec)
508 : {
509 0 : u32 status = 0;
510 0 : commVec = CreateCommP2PAsync(tag, inputMem, outputMem, commParaInfo.peerUserRank, status);
511 0 : for (u32 index = 0; index < commVec.size(); index++) {
512 0 : CHK_PRT_RET(!commVec[index],
513 : HCCL_ERROR("[Create][CommP2PSync]errNo[0x%016llx] tag[%s], created p2pComm[%u] is null.",
514 : HCCL_ERROR_CODE(HCCL_E_NOT_FOUND), tag.c_str(), index), HCCL_E_NOT_FOUND);
515 : }
516 :
517 0 : if (status == 0) {
518 0 : return HCCL_SUCCESS;
519 : }
520 : do {
521 0 : HcclResult ret = CreateCommP2PQuarry(commVec, status);
522 0 : if (ret != HCCL_SUCCESS) {
523 0 : HCCL_ERROR("[Create][CommP2P]comm p2p init failed");
524 0 : return ret;
525 : }
526 0 : SaluSleep(COMM_P2P_QUERRY_WAIT_TIME);
527 0 : } while (status == 1);
528 0 : return HCCL_SUCCESS;
529 : }
530 :
531 0 : std::vector<std::unique_ptr<CommBase> > CommFactory::CreateCommP2PAsync(const std::string &tag,
532 : const DeviceMem& inputMem, const DeviceMem& outputMem, const u32 dstUserRank, u32& status)
533 : {
534 0 : u32 ringSize = CommPlaneVector_[COMM_COMBINE].size();
535 0 : std::vector<std::unique_ptr<CommBase> > commP2PArray(0); // 复用CommBase来实现P2P拓扑功能
536 :
537 0 : bool memFlag = !inputMem || !outputMem;
538 0 : CHK_PRT_RET(memFlag, HCCL_ERROR("[Create][CommP2P]inputMem is null or outputMem is null."), commP2PArray);
539 :
540 0 : commP2PArray.resize(ringSize); // ring_size即为网络平面数, 比如能组几条环
541 :
542 0 : bool invalidcheck = (rankVector_.size() <= userRank_) || (rankVector_.size() <= dstUserRank);
543 0 : CHK_PRT_RET(invalidcheck, HCCL_ERROR("[Create][CommP2P]dstUserRank[%u] or userRank[%u] is bigger than rank vector.",
544 : dstUserRank, userRank_), commP2PArray);
545 :
546 0 : for (u32 ringIndex = 0; ringIndex < ringSize; ringIndex++) {
547 0 : u32 rank = GetSubCollectiveRank(CommPlaneVector_[COMM_COMBINE][ringIndex]);
548 0 : if (rank == INVALID_VALUE_RANKID) {
549 0 : continue;
550 : }
551 :
552 0 : IntraExchanger exchangerNetwork {};
553 0 : HCCL_INFO("[CreateCommP2PAsync] CommP2P is used %s. userRank = %u, rank = %u",
554 : isUsedRdmaLevel0_ ? "rdma" : "sdma", userRank_, rank);
555 0 : commP2PArray[ringIndex].reset(new (std::nothrow) CommP2P(identifier_, userRank_, userRankSize_,
556 0 : rank, CommPlaneVector_[COMM_COMBINE][ringIndex].size(), TopoType::TOPO_TYPE_COMMON, dispatcher_,
557 0 : notifyPool_, netDevCtxMap_, exchangerNetwork, CommPlaneVector_[COMM_COMBINE][ringIndex], inputMem,
558 0 : outputMem, isUsedRdmaLevel0_, tag, dstUserRank,
559 0 : nicDeployInner_));
560 :
561 0 : CHK_PRT_RET(!commP2PArray[ringIndex], HCCL_ERROR("[Create][CommP2P]comm p2p array[%u] reset failed.",
562 : ringIndex), commP2PArray);
563 0 : if (commP2PArray[ringIndex]->BuildAsync(status) != HCCL_SUCCESS) {
564 0 : HCCL_ERROR("[Create][CommP2P]comm p2p array[%u] init failed", ringIndex);
565 0 : commP2PArray[ringIndex].reset(nullptr);
566 0 : return commP2PArray;
567 : }
568 0 : HCCL_DEBUG("BuildAsync %u", status);
569 0 : }
570 0 : return commP2PArray;
571 0 : }
572 :
573 0 : HcclResult CommFactory::CreateCommP2PQuarry(std::vector<std::unique_ptr<CommBase> >& comm, u32& status)
574 : {
575 : HcclResult ret;
576 0 : std::vector<u32> commStatus(comm.size());
577 0 : for (u32 index = 0; index < comm.size(); index++) {
578 0 : CHK_SMART_PTR_NULL(comm[index]);
579 0 : ret = comm[index]->BuildQuerry(commStatus[index]);
580 0 : if (ret != HCCL_SUCCESS) {
581 0 : HCCL_ERROR("[Quarry][CommP2P]comm p2p array[%u] init failed", index);
582 0 : comm[index].reset(nullptr);
583 0 : return ret;
584 : }
585 : }
586 0 : status = (static_cast<int>(comm.size()) == std::count(commStatus.begin(), commStatus.end(), 0)) ? 0 : 1;
587 0 : HCCL_DEBUG("CreateCommP2PQuarry %u", status);
588 0 : return HCCL_SUCCESS;
589 0 : }
590 :
591 0 : u32 CommFactory::GetSubRootUserRank(const u32 userRank, const u32 rootUserRank)
592 : {
593 0 : u32 tmpUserRank = INVALID_VALUE_RANKID;
594 0 : if ((rankVector_.size() > userRank) && (rankVector_.size() > rootUserRank)) {
595 0 : u32 moduleIdx = 0;
596 0 : CHK_PRT_RET(topoInfoEx_->GetModuleIdx(rankVector_[rootUserRank], moduleIdx) != HCCL_SUCCESS,
597 : HCCL_ERROR("[Get][SubRootUserRank]get server id failed."), INVALID_VALUE_RANKID);
598 :
599 0 : auto iterRankRoot = serverToRank_.find(moduleIdx);
600 0 : CHK_PRT_RET(iterRankRoot == serverToRank_.end(),
601 : HCCL_ERROR("[Get][SubRootUserRank]can't find root serverId[%s] in rank map",
602 : rankVector_[rootUserRank].serverId.c_str()), INVALID_VALUE_RANKID);
603 :
604 0 : CHK_PRT_RET(topoInfoEx_->GetModuleIdx(rankVector_[userRank], moduleIdx) != HCCL_SUCCESS,
605 : HCCL_ERROR("[Get][SubRootUserRank]get server id failed."), INVALID_VALUE_RANKID);
606 :
607 0 : auto iterRankCurr = serverToRank_.find(moduleIdx);
608 0 : CHK_PRT_RET(iterRankCurr == serverToRank_.end(),
609 : HCCL_ERROR("[Get][SubRootUserRank]can't find local serverId[%s] in rank map",
610 : rankVector_[userRank].serverId.c_str()), INVALID_VALUE_RANKID);
611 :
612 0 : for (u32 index = 0; index < (iterRankCurr->second).size(); index++) {
613 : /* 当userRank的server内rank号与rootUserRank所在服务器中某一个server内rank号相同，
614 : 获取出rootUserRank所在服务器内的userrank */
615 0 : if (userRank == (iterRankCurr->second)[index].userRank) {
616 0 : tmpUserRank = (iterRankRoot->second)[index].userRank;
617 0 : break;
618 : }
619 : }
620 : }
621 0 : return tmpUserRank;
622 : }
623 :
624 0 : u32 CommFactory::GetSubRootUserRankWithSuperPod(const u32 userRank, const u32 rootUserRank)
625 : {
626 0 : u32 tmpUserRank = INVALID_VALUE_RANKID;
627 0 : if ((rankVector_.size() <= userRank) || (rankVector_.size() <= rootUserRank)) {
628 0 : return tmpUserRank;
629 : }
630 :
631 0 : u32 rootSuperPodIdx = rankVector_[rootUserRank].superPodIdx;
632 0 : auto iterRankRoot = superPodToRank_.find(rootSuperPodIdx);
633 0 : CHK_PRT_RET(iterRankRoot == superPodToRank_.end(),
634 : HCCL_ERROR("[Get][GetSubRootUserRankWithSuperPod]can't find root rootSuperPodIdx[%u] in rank map",
635 : rootSuperPodIdx), INVALID_VALUE_RANKID);
636 :
637 0 : u32 userSuperPodIdx = rankVector_[userRank].superPodIdx;
638 0 : auto iterRankCurr = superPodToRank_.find(userSuperPodIdx);
639 0 : CHK_PRT_RET(iterRankCurr == superPodToRank_.end(),
640 : HCCL_ERROR("[Get][GetSubRootUserRankWithSuperPod]can't find local userSuperPodIdx[%u] in rank map",
641 : userSuperPodIdx), INVALID_VALUE_RANKID);
642 :
643 0 : for (u32 index = 0; index < (iterRankCurr->second).size(); index++) {
644 : /* 当userRank的superPod内rank号与rootUserRank所在服务器中某一个superPod内rank号相同，
645 : 获取出rootUserRank所在服务器内的userrank */
646 0 : if (userRank == (iterRankCurr->second)[index].userRank) {
647 0 : tmpUserRank = (iterRankRoot->second)[index].userRank;
648 0 : break;
649 : }
650 : }
651 :
652 0 : return tmpUserRank;
653 : }
654 :
655 0 : u32 CommFactory::GetSubRootForScatter(const u32 root)
656 : {
657 : // 通过root找到ringIndex, 通过userRank找到level1中的rank
658 0 : u32 subRoot = INVALID_VALUE_RANKID;
659 0 : u32 planeIdx = INVALID_VALUE_RANKID;
660 0 : u32 ringSize = CommPlaneVector_[COMM_LEVEL1].size();
661 :
662 0 : CHK_PRT_RET(CommPlaneVector_[COMM_LEVEL1].size() == 0,
663 : HCCL_ERROR("[GET][GetSubRootForScatter]bridgeRankVector size is zero."), HCCL_E_PARA);
664 :
665 0 : u32 rank = INVALID_VALUE_RANKID;
666 0 : for (u32 ringIndex = 0; ringIndex < ringSize; ringIndex++) {
667 0 : if (isBridgeVector_[ringIndex]) {
668 0 : rank = GetSubCollectiveRank(CommPlaneVector_[COMM_LEVEL1][ringIndex]); // 确定userRank在level1中的rank号
669 : }
670 0 : for (u32 idx = 0; idx < CommPlaneVector_[COMM_LEVEL1][ringIndex].size(); idx++) {
671 0 : if (root == CommPlaneVector_[COMM_LEVEL1][ringIndex][idx].userRank) { // 获取root所在的平面
672 0 : planeIdx = ringIndex;
673 : }
674 : }
675 : }
676 0 : CHK_PRT_RET(rank == INVALID_VALUE_RANKID,
677 : HCCL_ERROR("[GET][GetSubRootForScatter]get rankId in level1 failed."), HCCL_E_PARA);
678 0 : CHK_PRT_RET(planeIdx == INVALID_VALUE_RANKID,
679 : HCCL_ERROR("[GET][GetSubRootForScatter]get root[%u] planeIdx[%u] failed.", root, planeIdx), HCCL_E_PARA);
680 0 : subRoot = CommPlaneVector_[COMM_LEVEL1][planeIdx][rank].userRank;
681 0 : HCCL_DEBUG("[GetSubRootForScatter] userRank_:[%u] subRoot:[%u]", userRank_, subRoot);
682 0 : return subRoot;
683 : }
684 :
685 10 : u32 CommFactory::GetSubCollectiveRank(const std::vector<RankInfo> &vecPara) const
686 : {
687 : // 在vecPara数据中，查询本user rank，查询到的vec下标就是rank值
688 10 : u32 tmpRank = INVALID_VALUE_RANKID;
689 :
690 10 : for (u32 rankIndex = 0; rankIndex < vecPara.size(); rankIndex++) {
691 10 : if (userRank_ == vecPara[rankIndex].userRank) {
692 10 : tmpRank = rankIndex;
693 10 : break;
694 : }
695 : }
696 :
697 10 : return tmpRank;
698 : }
699 :
700 0 : u32 CommFactory::GetLevel1CommRank(const u32 ringIdx)
701 : {
702 0 : return GetSubCollectiveRank(CommPlaneVector_[COMM_LEVEL1][ringIdx]);
703 : }
704 :
705 10 : bool CommFactory::JudgmentSetHeterogP2p(u32 rank) const
706 : {
707 10 : return isHaveCpuRank_;
708 : }
709 :
710 7 : HcclResult CommFactory::SetHDCModeInfo(
711 : std::unordered_map<std::string, std::map<u32, HcclIpAddress>> &rankDevicePhyIdNicInfoMap,
712 : std::vector<u32> &ranksPort, std::vector<u32> &vnicRanksPort, bool isSetHDCModeInfo, bool isUseRankPort)
713 : {
714 7 : rankDevicePhyIdNicInfoMap_ = rankDevicePhyIdNicInfoMap;
715 7 : ranksPort_ = ranksPort;
716 7 : vnicRanksPort_ = vnicRanksPort;
717 7 : isSetHDCModeInfo_ = isSetHDCModeInfo;
718 7 : isUseRankPort_ = isUseRankPort;
719 7 : return HCCL_SUCCESS;
720 : }
721 :
722 0 : HcclResult CommFactory::SetIsUsedRdma(const CommParaInfo &commParaInfo,
723 : std::vector<SingleSubCommTransport> &commTransport, bool isUsedRdma)
724 : {
725 0 : u32 ringSize = commTransport.size();
726 :
727 0 : for (u32 ringIndex = 0; ringIndex < ringSize; ringIndex++) {
728 0 : SingleSubCommTransport &subCommTransport = commTransport[ringIndex];
729 0 : for (auto &transportRequest : subCommTransport.transportRequests) {
730 0 : transportRequest.isUsedRdma = isUsedRdma;
731 : }
732 : }
733 0 : HCCL_INFO("[CommFactory][SetIsUsedRdma] commPlane[%d] isUsedRdma[%d]", commParaInfo.commPlane, isUsedRdma);
734 0 : return HCCL_SUCCESS;
735 : }
736 :
737 : } // namespace hccl
|
