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