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