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 "../rank_graph_builder/updater_for_64_plus_1.h"
12 : #include "topo_common_types.h"
13 : #include "exception_util.h"
14 : #include "invalid_params_exception.h"
15 :
16 : namespace Hccl {
17 :
18 : using namespace std;
19 :
20 54 : void UpdaterFor64Plus1::SaveReplaceInfo(const NewRankInfo& rank)
21 : {
22 54 : if (rank.localId != BACKUP_LOCAL_ID) {
23 51 : return;
24 : }
25 3 : string netInstId = "";
26 5 : for (const auto& netLayerInfo : rank.rankLevelInfos) {
27 2 : if (netLayerInfo.netLayer == 0) {
28 2 : netInstId = netLayerInfo.netInstId;
29 : }
30 : }
31 3 : if (netInstId.empty()) {
32 2 : THROW<InvalidParamsException>(StringFormat(
33 : "[UpdaterFor64Plus1::Init] "
34 : "Replaced rank[%d] has empty R0Id",
35 1 : rank.rankId));
36 : }
37 2 : if (replaceInfo.find(netInstId) != replaceInfo.end()) {
38 0 : THROW<InvalidParamsException>(StringFormat(
39 : "[UpdaterFor64Plus1::Init] "
40 : "R0Group[%s] has more than one backups",
41 : netInstId.c_str()));
42 : }
43 2 : replaceInfo[netInstId] = make_pair(rank.localId, rank.replacedLocalId);
44 3 : }
45 :
46 19 : void UpdaterFor64Plus1::UpdateRankGraph(RankGraph* rankGraph, const RankTableInfo* rankTable) const
47 : {
48 19 : if (rankGraph == nullptr) {
49 0 : THROW<NullPtrException>(StringFormat("[UpdaterFor64Plus1][%s] rankGraph is nullptr", __func__));
50 : }
51 19 : if (rankTable == nullptr) {
52 0 : THROW<NullPtrException>(StringFormat("[UpdaterFor64Plus1][%s] rankTable is nullptr", __func__));
53 : }
54 20 : for (const auto& it : replaceInfo) {
55 1 : const auto& netInstId = it.first;
56 1 : s32 localId = it.second.first;
57 1 : s32 replacedLocalId = it.second.second;
58 1 : NetInstance* netInstance = rankGraph->GetNetInstanceByNetInstId(0, netInstId);
59 1 : if (netInstance == nullptr) {
60 0 : HCCL_WARNING(
61 : "[UpdaterFor64Plus1][%s] netInstance netlayer[0] netInstanceId[%s] not exist", __func__,
62 : netInstId.c_str());
63 0 : continue;
64 0 : }
65 1 : UpdateNetInstance(netInstance, localId, replacedLocalId, rankTable);
66 : }
67 19 : }
68 :
69 1 : void UpdaterFor64Plus1::UpdateNetInstance(
70 : NetInstance* netInstance, LocalId localId, LocalId replacedLocalId, const RankTableInfo* rankTable) const
71 : {
72 1 : if (netInstance == nullptr) {
73 0 : THROW<NullPtrException>(StringFormat("[UpdaterFor64Plus1][%s] netInstance is nullptr", __func__));
74 : }
75 1 : auto phyTopoGraph = PhyTopo::GetInstance()->GetTopoGraph(0);
76 1 : if (phyTopoGraph == nullptr) {
77 0 : THROW<NullPtrException>(StringFormat("[UpdaterFor64Plus1][%s] phyTopoGraph is nullptr", __func__));
78 : }
79 3 : HCCL_DEBUG(
80 : "[UpdaterFor64Plus1][%s] Updating NetInstance[%s]: localId[%u]->replacedId[%u]", __func__,
81 : netInstance->GetNetInstId().c_str(), localId, replacedLocalId);
82 :
83 : // 找到与故障D直连的D
84 1 : shared_ptr<NetInstance::Peer> backupPeer; // local[64]
85 1 : vector<shared_ptr<NetInstance::Peer>> backupLinkedPeers;
86 5 : for (const auto& it : netInstance->GetPeers()) {
87 4 : auto peer = it.second;
88 4 : if (peer->GetLocalId() == BACKUP_LOCAL_ID) {
89 1 : backupPeer = peer;
90 1 : continue;
91 : }
92 3 : if (!IsSameX(peer->GetLocalId(), replacedLocalId) && !IsSameY(peer->GetLocalId(), replacedLocalId)) {
93 1 : continue;
94 : }
95 2 : if (!phyTopoGraph->HasEdge(PhyTopo::Peer::GetId(peer->GetLocalId()), PhyTopo::Peer::GetId(replacedLocalId))) {
96 0 : continue;
97 : }
98 2 : backupLinkedPeers.emplace_back(peer);
99 4 : }
100 :
101 : // 添加与故障D直连的D到备份D的Link
102 3 : for (const auto& peer : backupLinkedPeers) {
103 2 : AddPeer2BackupLinks(peer, backupPeer, replacedLocalId, netInstance, rankTable);
104 : }
105 1 : }
106 :
107 2 : void UpdaterFor64Plus1::AddPeer2BackupLinks(
108 : shared_ptr<NetInstance::Peer> peer, shared_ptr<NetInstance::Peer> backupPeer, LocalId replacedLocalId,
109 : NetInstance* netInstance, [[maybe_unused]] const RankTableInfo* rankTable) const
110 : {
111 2 : auto phyTopoGraph = PhyTopo::GetInstance()->GetTopoGraph(0);
112 :
113 2 : std::unordered_map<u64, u64> fabricIds;
114 : auto peer2AllPlaneEdges
115 2 : = phyTopoGraph->GetEdges(PhyTopo::Peer::GetId(backupPeer->GetLocalId()), PhyTopo::Fabric::GetId());
116 10 : for (auto edge : peer2AllPlaneEdges) {
117 8 : auto topoInstId = edge->GetTopoInstId();
118 8 : auto fabricId = static_cast<u64>(topoInstId) | (static_cast<u64>(1) << 32);
119 8 : fabricIds[topoInstId] = fabricId;
120 8 : }
121 :
122 2 : auto idx = GetLinkIndex(peer->GetLocalId(), replacedLocalId);
123 2 : auto backupPlaneId = idx.first;
124 2 : auto backupLinkIdx = idx.second;
125 6 : HCCL_DEBUG(
126 : "[UpdaterFor64Plus1][%s] Peer{rankId[%d], localId[%u]} will use BackupPlane[%u] addr[%u]", __func__,
127 : peer->GetRankId(), peer->GetLocalId(), backupPlaneId, backupLinkIdx);
128 :
129 : // 先获取phyTopoGraph中备份面和备份D的连接，因为只有一个fabric，所以会获取到全量16条备份面和备份D的连接
130 : // Edges中可能包含多个连接，但是在phytopo中保存为一条连接，内部有多个连接的端口
131 : // 拿到物理边后，根据backupPlaneId匹配，选择对应的一条物理边
132 2 : std::shared_ptr<PhyTopo::Link> backD2PlaneEdges = GetPeer2PlaneEdges(backupPlaneId, backupPeer, phyTopoGraph);
133 2 : std::shared_ptr<PhyTopo::Link> peer2PlaneEdges = GetPeer2PlaneEdges(backupPlaneId, peer, phyTopoGraph);
134 :
135 : // 取得端口列表，根据backupLinkIdx去选择对应的端口
136 2 : std::set<std::string> backD2PlanePorts = backD2PlaneEdges->GetSourceIFace()->GetPorts();
137 2 : std::set<std::string> peer2PlanePorts = peer2PlaneEdges->GetSourceIFace()->GetPorts();
138 :
139 : // 校验端口集合大小是否符合预期
140 2 : if (backD2PlanePorts.size() != BACKUP_TO_PLANE_ADDR_NUM) {
141 0 : THROW<InvalidParamsException>(
142 : "[UpdaterFor64Plus1][%s] Backup to BackupPlane[%u] port num error", __func__, backupPlaneId);
143 : }
144 2 : if (peer2PlanePorts.size() != 1) {
145 0 : THROW<InvalidParamsException>(
146 : "[UpdaterFor64Plus1][%s] Peer[%u] to BackupPlane[%u] port num error", __func__, peer->GetLocalId(),
147 : backupPlaneId);
148 : }
149 :
150 : // 从端口集合中取出对应逻辑位置的端口
151 2 : std::string backD2PlanePort = GetPortFromSet(backD2PlanePorts, backupLinkIdx);
152 2 : std::string peer2PlanePort = GetPortFromSet(peer2PlanePorts, 0);
153 :
154 : // 匹配了对应的端口后，去ranktableinfo中查对应端口的地址信息
155 : // todo 加一下peer->GetPortAddrMapLayer0()是否能找到port对应的地址。
156 2 : IpAddress backD2PlaneAddr = backupPeer->GetPortAddrMapLayer0()[backD2PlanePort][0];
157 2 : IpAddress peer2PlaneAddr = peer->GetPortAddrMapLayer0()[peer2PlanePort][0];
158 :
159 : // 组装成NetInstance的conninterface，加入peer和backupPeer
160 2 : if (backD2PlaneEdges->GetSourceIFace() == nullptr || peer2PlaneEdges->GetSourceIFace() == nullptr) {
161 0 : THROW<InvalidParamsException>("[UpdaterFor64Plus1][%s] source ConnInterface is nullptr", __func__);
162 : }
163 2 : std::set<string> peer2PlanePortSet;
164 2 : peer2PlanePortSet.insert(peer2PlanePort);
165 2 : std::set<string> backD2PlanePortSet;
166 2 : backD2PlanePortSet.insert(backD2PlanePort);
167 :
168 : shared_ptr<NetInstance::ConnInterface> backupIface = make_shared<NetInstance::ConnInterface>(
169 4 : backD2PlaneAddr, backD2PlanePortSet, backD2PlaneEdges->GetSourceIFace()->GetPos(), LinkType::PEER2PEER,
170 6 : backD2PlaneEdges->GetLinkProtocols(), backD2PlaneEdges->GetTopoType(), backD2PlaneEdges->GetTopoInstId());
171 : shared_ptr<NetInstance::ConnInterface> peerIface = make_shared<NetInstance::ConnInterface>(
172 4 : peer2PlaneAddr, peer2PlanePortSet, peer2PlaneEdges->GetSourceIFace()->GetPos(), LinkType::PEER2PEER,
173 6 : peer2PlaneEdges->GetLinkProtocols(), backD2PlaneEdges->GetTopoType(), backD2PlaneEdges->GetTopoInstId());
174 :
175 2 : backupPeer->AddConnInterface(0, backupIface);
176 2 : peer->AddConnInterface(0, peerIface);
177 :
178 : shared_ptr<NetInstance::Link> peer2Backup = make_shared<NetInstance::Link>(
179 2 : peer, backupPeer, peerIface, backupIface, LinkType::PEER2PEER, backD2PlaneEdges->GetLinkProtocols());
180 2 : netInstance->AddLink(peer2Backup);
181 : shared_ptr<NetInstance::Link> backup2Peer = make_shared<NetInstance::Link>(
182 2 : backupPeer, peer, backupIface, peerIface, LinkType::PEER2PEER, backD2PlaneEdges->GetLinkProtocols());
183 2 : netInstance->AddLink(backup2Peer);
184 :
185 : // peer2peer建立后删除graph中DB到所选planeId的对应的peer2net链路
186 : // 直接删除备份d和fabric的链接保证GetLinks接口只能获取到peer2db的一条peer2peer
187 : // 删除peer到fabric用到的peer2net
188 10 : for (auto id : fabricIds) {
189 8 : netInstance->DeleteLink(backupPeer->GetNodeId(), id.second);
190 : }
191 2 : }
192 :
193 4 : std::shared_ptr<PhyTopo::Link> UpdaterFor64Plus1::GetPeer2PlaneEdges(
194 : u32 backupPlaneId, shared_ptr<NetInstance::Peer> peer,
195 : std::shared_ptr<Graph<PhyTopo::Node, PhyTopo::Link>> phyTopoGraph) const
196 : {
197 4 : std::shared_ptr<PhyTopo::Link> peer2PlaneEdges = nullptr;
198 : auto peer2AllPlaneEdges
199 4 : = phyTopoGraph->GetEdges(PhyTopo::Peer::GetId(peer->GetLocalId()), PhyTopo::Fabric::GetId());
200 4 : if (peer2AllPlaneEdges.size() != BACKUP_PLANE_NUM) {
201 0 : THROW<InvalidParamsException>("[UpdaterFor64Plus1][%s] BackupPlane num error", __func__);
202 : }
203 8 : for (auto& edges : peer2AllPlaneEdges) {
204 8 : if (edges->GetTopoInstId() == backupPlaneId) {
205 4 : peer2PlaneEdges = edges;
206 4 : break;
207 : }
208 : }
209 4 : if (peer2PlaneEdges == nullptr) {
210 0 : THROW<NullPtrException>(StringFormat("[UpdaterFor64Plus1][%s] peer2PlaneEdges is nullptr", __func__));
211 : }
212 4 : return peer2PlaneEdges;
213 4 : }
214 :
215 7 : std::string UpdaterFor64Plus1::GetPortFromSet(std::set<string>& ports, u32 linkIdx) const
216 : {
217 7 : std::string peer2PlanePort = "";
218 7 : if (linkIdx >= ports.size()) {
219 1 : THROW<InvalidParamsException>("[UpdaterFor64Plus1][%s] BackupPlane port num error", __func__);
220 : }
221 9 : for (auto& port : ports) {
222 9 : if (linkIdx == 0) {
223 6 : peer2PlanePort = port;
224 6 : break;
225 : }
226 3 : linkIdx--;
227 : }
228 6 : return peer2PlanePort;
229 1 : }
230 :
231 10 : bool UpdaterFor64Plus1::IsSameX(LocalId srcLocalId, LocalId dstLocalId) const
232 : {
233 : // 两个localId对应的D是否在同一个X轴
234 10 : return (srcLocalId / DEVICE_NUM_PER_AXIS) == (dstLocalId / DEVICE_NUM_PER_AXIS);
235 : }
236 :
237 6 : bool UpdaterFor64Plus1::IsSameY(LocalId srcLocalId, LocalId dstLocalId) const
238 : {
239 : // 两个localId对应的D是否在同一个Y轴
240 6 : return (srcLocalId % DEVICE_NUM_PER_AXIS) == (dstLocalId % DEVICE_NUM_PER_AXIS);
241 : }
242 :
243 9 : pair<u32, u32> UpdaterFor64Plus1::GetLinkIndex(LocalId localId, LocalId replacedLocalId) const
244 : {
245 9 : if (localId >= BACKUP_LOCAL_ID || replacedLocalId >= BACKUP_LOCAL_ID || localId == replacedLocalId) {
246 2 : THROW<InvalidParamsException>(
247 : "[UpdaterFor64Plus1][%s] localId[%u] or replacedId[%u] invalid", __func__, localId, replacedLocalId);
248 : return {};
249 : }
250 7 : if (IsSameX(localId, replacedLocalId)) {
251 3 : auto idx = localId % DEVICE_NUM_PER_AXIS;
252 3 : if (idx < DEVICE_HALF_NUM_PER_AXIS) {
253 2 : return make_pair(0, idx); // X轴左边4个正常D走备份面0
254 : } else {
255 1 : return make_pair(1, idx % DEVICE_HALF_NUM_PER_AXIS); // X轴右边4个正常D走备份面1
256 : }
257 4 : } else if (IsSameY(localId, replacedLocalId)) {
258 3 : auto idx = localId / DEVICE_NUM_PER_AXIS;
259 3 : if (idx < DEVICE_HALF_NUM_PER_AXIS) {
260 2 : return make_pair(2, idx); // Y轴上边4个正常D走备份面2
261 : } else {
262 1 : return make_pair(3, idx % DEVICE_HALF_NUM_PER_AXIS); // Y轴下边4个正常D走备份面3
263 : }
264 : } else {
265 1 : THROW<InvalidParamsException>(
266 : "[UpdaterFor64Plus1][%s] localId[%u] and replacedId[%u] are not at same line", __func__, localId,
267 : replacedLocalId);
268 : return {};
269 : }
270 : }
271 :
272 : } // namespace Hccl
|
