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 "topoinfo_exchange_server.h"
12 : #include <thread>
13 : #include <fstream>
14 : #include <iostream>
15 : #include "externalinput_pub.h"
16 : #include "config.h"
17 : #include "hccl_socket.h"
18 : #include "sal_pub.h"
19 : #include "topoinfo_exchange_dispatcher.h"
20 : #include "preempt_port_manager.h"
21 :
22 : namespace hccl {
23 : const u32 DISPLAY_RANKNUM_PERLINE = 8;
24 : const u32 SOCKET_ACCEPT_TIMEOUT = 60; //Server调用Accept等待的最大超时时间 60s
25 : const u32 SOCKET_PRINT_COUNT = 3; //未建链打印的数量
26 : using namespace std;
27 17 : TopoInfoExchangeServer::TopoInfoExchangeServer(HcclIpAddress &hostIP, u32 hostPort,
28 : const std::vector<HcclIpAddress> whitelist, HcclNetDevCtx netDevCtx,
29 17 : std::shared_ptr<HcclSocket> listenSocket, const std::string &identifier)
30 17 : : hostIP_(hostIP),
31 17 : hostPort_(hostPort),
32 17 : whitelist_(whitelist),
33 17 : netDevCtx_(netDevCtx),
34 17 : listenSocket_(listenSocket),
35 34 : identifier_(identifier)
36 : {
37 17 : }
38 :
39 0 : TopoInfoExchangeServer::TopoInfoExchangeServer(HcclIpAddress &hostIP, u32 hostPort,
40 : const std::vector<HcclIpAddress> whitelist, HcclNetDevCtx netDevCtx, std::shared_ptr<HcclSocket> listenSocket,
41 0 : std::shared_ptr<HcclSocket> grpLeaderToRoot, const std::string &identifier)
42 0 : : hostIP_(hostIP),
43 0 : hostPort_(hostPort),
44 0 : whitelist_(whitelist),
45 0 : netDevCtx_(netDevCtx),
46 0 : listenSocket_(listenSocket),
47 0 : grpLeaderToRoot_(grpLeaderToRoot),
48 0 : identifier_(identifier)
49 : {
50 0 : }
51 :
52 30 : TopoInfoExchangeServer::~TopoInfoExchangeServer()
53 : {
54 30 : }
55 :
56 13 : HcclResult TopoInfoExchangeServer::FailedConnectionAgentIdString(u32 rankSize, std::string &failedAgentIdList)
57 : {
58 13 : HcclResult result = HCCL_E_NOT_FOUND;
59 13 : const u32 oriLength = failedAgentIdList.length();
60 13 : std::vector<bool> connectedRank(rankSize, false);
61 13 : for (auto it : connectSocketsWithRankID_) {
62 0 : if (it.first >= rankSize) {
63 0 : HCCL_ERROR("[TopoInfoExchangeServer][FailedConnectionAgentIdString] invalid rank id[%u] from agent.",
64 : it.first);
65 0 : return HCCL_E_INTERNAL;
66 : }
67 0 : connectedRank[it.first] = true;
68 0 : }
69 :
70 13 : for (u32 i = 0; i < rankSize; i++) {
71 0 : if (!connectedRank[i]) {
72 0 : failedAgentIdList += std::to_string(i) + ',';
73 : }
74 : }
75 :
76 13 : return failedAgentIdList.length() > oriLength ? HCCL_SUCCESS : result;
77 13 : }
78 :
79 13 : HcclResult TopoInfoExchangeServer::Setup()
80 : {
81 : HcclResult ret;
82 13 : HcclResult error = HCCL_SUCCESS;
83 :
84 : do {
85 13 : u32 expectRankSize = 0;
86 13 : std::string failedAgentIdList;
87 13 : HcclResult connectRet = Connect(connectSockets_, expectRankSize);
88 13 : if (connectRet != HCCL_SUCCESS) {
89 13 : HcclResult result = FailedConnectionAgentIdString(expectRankSize, failedAgentIdList);
90 13 : CHK_PRT_CONT(result == HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeServer]failed to connect rankList:[%s]", failedAgentIdList.c_str()));
91 : }
92 13 : u32 rankSize = connectSockets_.size();
93 13 : if (!isByMasterInfo_ && rankSize > TOPO_HIERARCHICAL_ENABLE_THRESHOLD) {
94 0 : ret = HierarchicalSendRecv();
95 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
96 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]HierarchicalSendRecv ranktable failed"), error = ret);
97 0 : HCCL_INFO("cluster topo exchange server HierarchicalSendRecv ranktable success.");
98 : } else {
99 13 : RankTable_t rankTable;
100 13 : ret = GetRanksBasicInfo(connectSockets_, rankTable);
101 13 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeServer][Setup]GetRanksBasicInfo failed"),
102 : error = ret);
103 13 : HCCL_INFO("cluster topo exchange server get rank basic info from all agent success.");
104 :
105 13 : g_broadcastStage.store(BroadcastStage::Started, std::memory_order_release);
106 13 : TopoInfoExchangeDispather dispatcher(this);
107 13 : ret = dispatcher.BroadcastRankTable(connectSockets_, rankTable, failedAgentIdList);
108 : {
109 13 : g_broadcastStage.store(BroadcastStage::Completed, std::memory_order_release);
110 13 : std::lock_guard<std::mutex> lock(g_broadcast_stage_mutex);
111 13 : g_broadcast_stage_cv.notify_all();
112 13 : }
113 13 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
114 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]Broadcast Rank Basic Infos failed, connectFailedAgentIdList[%s]", failedAgentIdList.c_str()),
115 : error = ret);
116 13 : HCCL_INFO("cluster topo exchange server send rank basic info to all agent success.");
117 13 : CHK_PRT_BREAK(connectRet != HCCL_SUCCESS,
118 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]cluster topo exchange server connect client failed"),
119 : error = connectRet);
120 0 : HCCL_INFO("cluster topo exchange server connect with all agent success.");
121 26 : }
122 0 : ret = StopSocketListen(whitelist_, hostPort_);
123 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
124 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]topo exchange server stop socket listen port[%u] failed.",
125 : hostPort_), error = ret);
126 13 : } while (0);
127 13 : if (error != HCCL_SUCCESS) {
128 13 : CHK_RET(Disconnect(connectSockets_));
129 13 : CHK_RET(StopNetwork(whitelist_, hostPort_));
130 : }
131 :
132 13 : HCCL_INFO("cluster topo exchange server completed, exit[%u].", error);
133 13 : return error;
134 : }
135 :
136 0 : HcclResult TopoInfoExchangeServer::HierarchicalSendRecv()
137 : {
138 0 : TopoInfoExchangeDispather dispatcherGrpLeader(this);
139 0 : TopoInfoExchangeDispather dispatcherGrpLeaderPortInfo(this);
140 0 : TopoInfoExchangeDispather dispatcherRankTable(this);
141 :
142 : // get Group Leader info
143 0 : GroupLeader_t groupLeader;
144 0 : HcclResult ret = RecvGroupLeaderInfo(connectSockets_, groupLeader);
145 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeServer][Setup]RecvGroupLeaderInfo failed"), ret);
146 :
147 0 : HCCL_INFO("cluster topo exchange server get group leader info.");
148 : // BroadCast GroupLeader info
149 0 : ret = dispatcherGrpLeader.BroadcastGroupLeaderInfo(connectSockets_, groupLeader);
150 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
151 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]Broadcast Group Leader Infos No PortInfo failed"), ret);
152 0 : HCCL_INFO("cluster topo exchange server send groupleader info to all agent success.");
153 :
154 : // root接收每个GroupLeader传上来的port
155 0 : ret = RecvGroupLeaderPortInfo(grpLeaderSockets_,groupLeader);
156 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeServer][Setup]RecvGroupLeaderPortInfo failed"), ret);
157 :
158 : // BroadCast GroupLeader Port Info
159 0 : ret = dispatcherGrpLeaderPortInfo.BroadcastGroupLeaderInfo(connectSockets_,groupLeader);
160 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
161 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]Broadcast Group Leader Infos with PortInfo failed"), ret);
162 0 : HCCL_INFO("cluster topo exchange server send groupleader info to all agent success.");
163 : // root接收GroupLeader上传的ranktable
164 0 : RankTable_t rankTable;
165 :
166 0 : ret = GetRanksBasicInfo(grpLeaderSockets_, rankTable);
167 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeServer][Setup]RecvGroupClusterInfo failed"), ret);
168 0 : HCCL_INFO("cluster topo exchange server get rank basic info from all group leader success.");
169 :
170 : // root向GroupLeader广播全局ranktable
171 0 : ret = dispatcherRankTable.BroadcastRankTable(grpLeaderSockets_, rankTable, "");
172 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
173 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]Broadcast Rank Basic Infos failed"), ret);
174 0 : HCCL_INFO("cluster topo exchange server send rank basic info to all group leader success.");
175 :
176 0 : return HCCL_SUCCESS;
177 0 : }
178 :
179 0 : HcclResult TopoInfoExchangeServer::RecvGroupLeaderInfo(
180 : const std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, GroupLeader_t &groupLeader)
181 : {
182 0 : u32 socketNumPerGrp = 0;
183 0 : u32 socketIndex = 0; // socket已经经过rankid（or superPodId + serverip + deviceid排序）
184 0 : bool isGroupLeader = true;
185 0 : std::map<u32, HcclRootHandle> GroupLeaders;
186 :
187 0 : for (auto &handle : connectSockets) {
188 : HcclRankHandle rankHandle;
189 0 : HcclResult ret = handle.second->Recv(&rankHandle, sizeof(HcclRankHandle));
190 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
191 : HCCL_ERROR("[Get][RecvGroupLeaderInfo]RecvGroupLeaderInfo from agentId[%s] failed, ret[%d]",
192 : handle.first.c_str(), ret), ret);
193 0 : if(isGroupLeader) {
194 0 : u32 GroupIndex = socketIndex / TOPO_MAX_GROUP_SIZE;
195 0 : GroupLeaders.insert(pair<u32, HcclRootHandle>(GroupIndex, rankHandle));
196 0 : grpLeaderSockets_.insert(handle);
197 0 : isGroupLeader = false;
198 : }
199 :
200 0 : socketNumPerGrp++;
201 0 : socketIndex++;
202 0 : if (socketNumPerGrp == TOPO_MAX_GROUP_SIZE) {
203 0 : isGroupLeader = true;
204 0 : socketNumPerGrp = 0;
205 : }
206 : }
207 : // 把GroupLeader信息存放到GroupLeaderList中 方便广播
208 0 : for (auto iter : GroupLeaders) {
209 0 : groupLeader.grpLeaderNum++;
210 0 : groupLeader.GroupLeaderList.emplace_back(iter.second);
211 : }
212 :
213 0 : return HCCL_SUCCESS;
214 0 : }
215 :
216 0 : HcclResult TopoInfoExchangeServer::RecvGroupLeaderPortInfo(
217 : const std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, GroupLeader_t &groupLeader)
218 : {
219 : HcclResult ret;
220 0 : groupLeader.GroupLeaderList.clear();
221 0 : for(auto &handle : connectSockets) {
222 : HcclRankHandle grpLeaderPortInfo;
223 0 : ret = handle.second->Recv(&grpLeaderPortInfo, sizeof(HcclRankHandle));
224 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
225 : HCCL_ERROR("[Get][RecvGroupLeaderPortInfo]RecvGroupLeaderPortInfo from grpLeader[%s] failed, ret[%d]",
226 : handle.first.c_str(), ret), ret);
227 0 : groupLeader.GroupLeaderList.emplace_back(grpLeaderPortInfo);
228 : }
229 0 : return HCCL_SUCCESS;
230 : }
231 :
232 0 : HcclResult TopoInfoExchangeServer::SetupGroupLeader()
233 : {
234 : HcclResult ret;
235 0 : HcclResult error = HCCL_SUCCESS;
236 :
237 : do {
238 0 : TopoInfoExchangeDispather dispatcher(this);
239 :
240 0 : ret = GroupLeaderConnect(connectSockets_);
241 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
242 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]cluster topo exchange server connect client failed"),
243 : error = ret);
244 0 : HCCL_INFO("cluster topo exchange server connect with all agent success.");
245 :
246 0 : RankTable_t rankTable;
247 : // GroupLeader接收Group内rank上报的ranktable
248 0 : ret = GetRanksBasicInfo(connectSockets_, rankTable);
249 0 : currentStep_--;
250 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS, HCCL_ERROR("[TopoInfoExchangeServer][Setup]RecvGroupClusterInfo failed"),
251 : error = ret);
252 0 : HCCL_INFO("cluster topo exchange server get rank basic info from all agent success.");
253 :
254 0 : HCCL_INFO("topo exchange client send rank basic info success.");
255 0 : CHK_RET(SendClusterInfo(grpLeaderToRoot_, rankTable));
256 :
257 0 : CHK_RET(RecvClusterInfo(grpLeaderToRoot_, rankTable_));
258 0 : currentStep_--;
259 0 : HCCL_INFO("topo exchange client get rank basic info success.");
260 :
261 0 : ret = dispatcher.BroadcastRankTable(connectSockets_, rankTable_, "");
262 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
263 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]Broadcast Rank Basic Infos failed"), error = ret);
264 0 : HCCL_INFO("cluster topo exchange server send rank basic info to all agent success.");
265 :
266 0 : ret = StopSocketListen(whitelist_, hostPort_);
267 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
268 : HCCL_ERROR("[TopoInfoExchangeServer][Setup]topo exchange server stop socket listen failed."), error = ret);
269 0 : } while (0);
270 :
271 0 : if (error != HCCL_SUCCESS) {
272 0 : CHK_RET(Disconnect(connectSockets_));
273 0 : CHK_RET(StopNetwork(whitelist_, hostPort_));
274 : }
275 :
276 0 : HCCL_INFO("cluster topo exchange server completed, exit[%u].", error);
277 :
278 0 : return error;
279 : }
280 :
281 1 : HcclResult TopoInfoExchangeServer::Teardown()
282 : {
283 1 : CHK_RET(Disconnect(connectSockets_));
284 1 : CHK_RET(StopNetwork(whitelist_, hostPort_));
285 1 : return HCCL_SUCCESS;
286 : }
287 :
288 1 : HcclResult TopoInfoExchangeServer::GetConnections(std::map<u32, std::shared_ptr<HcclSocket>> &connectSockets)
289 : {
290 1 : connectSockets = connectSocketsWithRankID_;
291 1 : return HCCL_SUCCESS;
292 : }
293 :
294 :
295 0 : HcclResult TopoInfoExchangeServer::SetupByMasterInfo()
296 : {
297 0 : isByMasterInfo_ = true;
298 0 : CHK_RET(Setup());
299 0 : return HCCL_SUCCESS;
300 : }
301 :
302 13 : HcclResult TopoInfoExchangeServer::Connect(std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, u32 &rankSize)
303 : {
304 13 : auto startTime = std::chrono::steady_clock::now();
305 13 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
306 13 : u32 expectSocketNum = 1;
307 13 : u32 previousRankNum = 0;
308 13 : bool isFirstAcceptTimeOut = false;
309 :
310 185978392 : while (expectSocketNum > 0) {
311 185978392 : auto topoExUsedTime = std::chrono::steady_clock::now() - startTime;
312 185978392 : if (topoExUsedTime >= timeout) {
313 13 : HCCL_ERROR("[%s][%s]topo exchange server get socket timeout! timeout[%d s]",
314 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_RANKTABLE_DETECT.c_str(), GetExternalInputHcclLinkTimeOut());
315 13 : DisplayConnectedRank(connectSockets, rankSize);
316 13 : return HCCL_E_TIMEOUT;
317 : }
318 185978379 : auto topoExResTime = timeout - topoExUsedTime;
319 185978379 : u32 topoExRes_i = std::chrono::duration_cast<std::chrono::seconds>(topoExResTime).count();
320 185978379 : u32 socketWaitTime = SOCKET_ACCEPT_TIMEOUT;
321 185978379 : if (topoExRes_i != 0) {
322 0 : socketWaitTime = topoExRes_i > SOCKET_ACCEPT_TIMEOUT ? SOCKET_ACCEPT_TIMEOUT : topoExRes_i;
323 : } else {
324 185978379 : continue;
325 : }
326 0 : std::shared_ptr<HcclSocket> socket;
327 0 : std::string tag = TOPO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(hostPort_);
328 0 : HcclResult ret = listenSocket_->Accept(tag, socket, socketWaitTime);
329 0 : if (ret == HCCL_SUCCESS) {
330 0 : HCCL_INFO("listenSocket_->Accept completed.");
331 : // server获取socket之后进行一次数据收发用于判断是否都成功获取到了socket
332 0 : CHK_RET(socket->Send(TOPO_EXCHANGE_CHECK_MESSAGE, sizeof(TOPO_EXCHANGE_CHECK_MESSAGE)));
333 0 : u32 rankNum = 0;
334 0 : CHK_RET(GetRemoteFdAndRankSize(socket, connectSockets, rankNum));
335 0 : rankSize = rankNum;
336 0 : expectSocketNum = (previousRankNum == 0) ? rankNum : expectSocketNum;
337 0 : CHK_RET(VerifyRemoteRankNum(previousRankNum, rankNum));
338 :
339 0 : expectSocketNum -= 1;
340 0 : isFirstAcceptTimeOut = false;
341 0 : } else if (ret == HCCL_E_TIMEOUT) {
342 0 : HCCL_INFO("listenSocket_->Accept TimeOut[%lld s]", socketWaitTime);
343 0 : if (isFirstAcceptTimeOut) {
344 0 : continue;
345 : }
346 0 : isFirstAcceptTimeOut = true;
347 :
348 0 : DisplayConnectingStatus(previousRankNum, expectSocketNum, connectSockets);
349 0 : } else if (ret == HCCL_E_TCP_CONNECT) {
350 0 : HCCL_INFO("listenSocket_->Accept E_TCP_CONNECT");
351 0 : DisplayConnectedRank(connectSockets, rankSize);
352 0 : return HCCL_E_TCP_CONNECT;
353 : }
354 0 : }
355 0 : return HCCL_SUCCESS;
356 : }
357 :
358 0 : HcclResult TopoInfoExchangeServer::GroupLeaderConnect(std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets)
359 : {
360 0 : auto startTime = std::chrono::steady_clock::now();
361 0 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
362 :
363 0 : u32 groupMaxRankNum = TOPO_MAX_GROUP_SIZE;
364 0 : bool isFirstAcceptTimeOut = false;
365 :
366 0 : while (expectSocketNum_ > 0 && groupMaxRankNum > 0) {
367 0 : auto topoExUsedTime = std::chrono::steady_clock::now() - startTime;
368 0 : if (topoExUsedTime >= timeout) {
369 0 : HCCL_ERROR("[%s][%s]topo exchange server get socket timeout! timeout[%d s]",
370 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_RANKTABLE_DETECT.c_str(), GetExternalInputHcclLinkTimeOut());
371 0 : DisplayConnectedRank(connectSockets);
372 0 : return HCCL_E_TIMEOUT;
373 : }
374 0 : auto topoExResTime = timeout - topoExUsedTime;
375 0 : u32 topoExRes_i = std::chrono::duration_cast<std::chrono::seconds>(topoExResTime).count();
376 0 : u32 socketWaitTime = SOCKET_ACCEPT_TIMEOUT;
377 0 : if (topoExRes_i != 0) {
378 0 : socketWaitTime = topoExRes_i > SOCKET_ACCEPT_TIMEOUT ? SOCKET_ACCEPT_TIMEOUT : topoExRes_i;
379 : } else {
380 0 : continue;
381 : }
382 0 : std::shared_ptr<HcclSocket> socket;
383 0 : std::string tag = TOPO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(hostPort_);
384 :
385 0 : HcclResult ret = listenSocket_->Accept(tag, socket, socketWaitTime);
386 0 : if (ret == HCCL_SUCCESS) {
387 0 : HCCL_INFO("listenSocket_->Accept completed.");
388 0 : u32 rankNum = 0;
389 0 : CHK_RET(GetRemoteFdAndRankSize(socket, connectSockets, rankNum));
390 0 : expectSocketNum_ = (previousRankNum_ == 0) ? rankNum : expectSocketNum_;
391 0 : groupMaxRankNum = (rankNum > TOPO_HIERARCHICAL_ENABLE_THRESHOLD) ?
392 : groupMaxRankNum : expectSocketNum_;
393 0 : CHK_RET(VerifyRemoteRankNum(previousRankNum_, rankNum));
394 :
395 0 : expectSocketNum_ -= 1;
396 0 : groupMaxRankNum -= 1;
397 0 : isFirstAcceptTimeOut = false;
398 0 : } else if (ret == HCCL_E_TIMEOUT) {
399 0 : HCCL_ERROR("listenSocket_->Accept TimeOut[%lld s]", socketWaitTime);
400 0 : if (isFirstAcceptTimeOut) {
401 0 : continue;
402 : }
403 0 : isFirstAcceptTimeOut = true;
404 :
405 0 : DisplayConnectingStatus(previousRankNum_, expectSocketNum_, connectSockets);
406 0 : } else if (ret == HCCL_E_TCP_CONNECT) {
407 0 : HCCL_INFO("listenSocket_->Accept E_TCP_CONNECT");
408 0 : DisplayConnectedRank(connectSockets);
409 0 : return HCCL_E_TCP_CONNECT;
410 : }
411 0 : }
412 :
413 0 : return HCCL_SUCCESS;
414 : }
415 :
416 0 : HcclResult TopoInfoExchangeServer::DisplayConnectingStatus(u32 totalSockets, u32 waitSockets,
417 : const std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets)
418 : {
419 0 : if (totalSockets == 0 && waitSockets == 1) {
420 0 : return HCCL_SUCCESS;
421 : }
422 :
423 : // 单算子模式阶段性打印内容
424 0 : if (!isByMasterInfo_) {
425 0 : std::vector<bool> rankinfos(totalSockets, false);
426 0 : for (auto it : connectSockets) { //建立映射
427 0 : u32 rankid = 0;
428 0 : CHK_RET(SalStrToULong(it.first, HCCL_BASE_DECIMAL, rankid));
429 0 : rankinfos.at(rankid) = true;
430 0 : }
431 :
432 0 : u32 unRankCount = 0;// 只打印前三条未建链的rank
433 0 : std::vector<string> unsocketinfos;
434 0 : for (u32 rankid = 0 ; rankid < totalSockets; rankid++) {
435 0 : if (unRankCount >= SOCKET_PRINT_COUNT) {
436 0 : break;
437 : }
438 0 : if (!rankinfos[rankid]) {
439 0 : unRankCount++;
440 0 : std::string rankID = std::to_string(rankid);
441 0 : std::string agentID = std::string(16 - rankID.length(), '0') + rankID;
442 0 : unsocketinfos.push_back(agentID);
443 0 : }
444 : }
445 :
446 0 : std::string infoStr = "succ sockets is [" + std::to_string((totalSockets - waitSockets)) +
447 0 : "], waiting sockets is [" + std::to_string(waitSockets) + "], wait sockets rankid: ";
448 0 : for (u32 index = 0; index < unsocketinfos.size(); index++) {
449 0 : if (index == (unsocketinfos.size()-1)) {
450 0 : infoStr += "["+ unsocketinfos[index] +"]";
451 : } else {
452 0 : infoStr += "["+ unsocketinfos[index] +"],";
453 : }
454 : }
455 :
456 0 : HCCL_RUN_INFO("[HCCL_TRACE] %s", infoStr.c_str());
457 0 : } else {
458 0 : std::string infoStr = "succ sockets is [" + std::to_string(totalSockets - waitSockets) +
459 0 : "], waiting sockets is [" + std::to_string(waitSockets) + "]";
460 0 : HCCL_RUN_INFO("[HCCL_TRACE] %s , isByMasterInfo[%d]", infoStr.c_str(), isByMasterInfo_);
461 0 : }
462 :
463 0 : return HCCL_SUCCESS;
464 : }
465 :
466 0 : HcclResult TopoInfoExchangeServer::GetRemoteFdAndRankSize(std::shared_ptr<HcclSocket> &socket,
467 : std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, u32 &rankSize)
468 : {
469 0 : std::string agentID;
470 0 : CHK_RET(RecvRemoteAgentID(socket, agentID));
471 0 : auto iter = connectSockets.find(agentID);
472 0 : CHK_PRT_RET(iter != connectSockets.end(),
473 : HCCL_ERROR("[Get][Connection]GetConnection failed. agnet[%s] has been connected.", agentID.c_str()),
474 : HCCL_E_INTERNAL);
475 0 : connectSockets.insert({ agentID, socket });
476 :
477 0 : CHK_RET(RecvRemoteRankNum(socket, rankSize));
478 :
479 0 : u32 rankID = 0;
480 0 : if (!isByMasterInfo_) {
481 0 : CHK_RET(SalStrToULong(agentID, HCCL_BASE_DECIMAL, rankID));
482 0 : connectSocketsWithRankID_.insert({rankID, socket});
483 : }
484 :
485 0 : bool isRankIdUnAvailable = isByMasterInfo_ ? (false) : (rankID >= rankSize);
486 0 : CHK_PRT_RET(isRankIdUnAvailable, HCCL_ERROR("[Get][Connection]rank"
487 : " num[%u] from remote[%s] invalid.", rankSize, agentID.c_str()), HCCL_E_INTERNAL);
488 0 : HCCL_INFO("get remote rank[%s / %u] success.", agentID.c_str(), rankSize);
489 0 : return HCCL_SUCCESS;
490 0 : }
491 :
492 13 : HcclResult TopoInfoExchangeServer::DisplayConnectedRank(
493 : const std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, u32 rankNum)
494 : {
495 13 : vector<string> ranksInfo;
496 13 : for (auto it : connectSockets) {
497 0 : ranksInfo.push_back(it.first);
498 0 : }
499 13 : u64 ranksLen = ranksInfo.size();
500 13 : u64 lineNum = (ranksInfo.size() % DISPLAY_RANKNUM_PERLINE == 0) ? (ranksInfo.size()/DISPLAY_RANKNUM_PERLINE) :
501 0 : (ranksInfo.size()/DISPLAY_RANKNUM_PERLINE + 1);
502 13 : HCCL_ERROR("[%s][%s]total connected num is [%llu],line num is [%llu]",
503 : LOG_KEYWORDS_INIT_GROUP.c_str(), __func__, ranksLen, lineNum);
504 13 : if (rankNum != 0) {
505 0 : HCCL_ERROR("[%s][%s]need connect rankNum is [%u]", LOG_KEYWORDS_INIT_GROUP.c_str(), __func__, rankNum);
506 : }
507 13 : for (u64 i = 0; i < lineNum; i++) {
508 0 : string tmpRankList;
509 0 : for (u32 j = 0; j < DISPLAY_RANKNUM_PERLINE; j++) {
510 0 : u32 ranksInfoIndex = i * DISPLAY_RANKNUM_PERLINE + j;
511 0 : if (ranksInfoIndex < ranksInfo.size()) {
512 0 : tmpRankList += "[" + ranksInfo[ranksInfoIndex] + "]";
513 : } else {
514 0 : break;
515 : }
516 0 : tmpRankList += ((j == DISPLAY_RANKNUM_PERLINE - 1 || ranksInfoIndex == ranksInfo.size() - 1) ? ";" : ",");
517 : }
518 0 : HCCL_ERROR("[%s][%s]connected rankinfo[LINE %llu]: %s", LOG_KEYWORDS_INIT_GROUP.c_str(),
519 : __func__, i, tmpRankList.c_str());
520 0 : }
521 13 : return HCCL_SUCCESS;
522 13 : }
523 :
524 14 : HcclResult TopoInfoExchangeServer::Disconnect(std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets)
525 : {
526 14 : std::unique_lock<std::mutex> lock(lock_);
527 14 : for (auto &socket : connectSockets) {
528 0 : CHK_RET(DisconnectSocket(socket.second));
529 : }
530 14 : connectSockets.clear();
531 14 : connectSocketsWithRankID_.clear();
532 14 : return HCCL_SUCCESS;
533 14 : }
534 :
535 1 : HcclResult TopoInfoExchangeServer::DeleteSocketWhiteList(u32 port,
536 : const std::vector<HcclIpAddress> &whitelist)
537 : {
538 1 : std::vector<SocketWlistInfo> wlistInfosVec;
539 2 : for (auto ip : whitelist) {
540 1 : SocketWlistInfo wlistInfo = {0};
541 1 : wlistInfo.connLimit = HOST_SOCKET_CONN_LIMIT;
542 1 : wlistInfo.remoteIp.addr = ip.GetBinaryAddress().addr;
543 1 : wlistInfo.remoteIp.addr6 = ip.GetBinaryAddress().addr6;
544 1 : std::string tag = TOPO_DETECT_TAG + "_" + identifier_ + "_" + std::to_string(port);
545 1 : s32 sRet = memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag), tag.c_str(), tag.size() + 1);
546 1 : if (sRet != EOK) {
547 0 : HCCL_ERROR("[Delete][SocketWhiteList]memory copy failed. errorno[%d]", sRet);
548 0 : return HCCL_E_MEMORY;
549 : }
550 1 : wlistInfosVec.push_back(wlistInfo);
551 1 : }
552 :
553 1 : listenSocket_->DelWhiteList(wlistInfosVec);
554 :
555 1 : HCCL_INFO("delete socket white list success. total: %zu", whitelist.size());
556 1 : return HCCL_SUCCESS;
557 1 : }
558 :
559 16 : HcclResult TopoInfoExchangeServer::StopSocketListen(const std::vector<HcclIpAddress> &whitelist,
560 : u32 hostPort)
561 : {
562 16 : if (listenSocket_) {
563 14 : if (GetExternalInputHcclEnableWhitelist() == HCCL_WHITELIST_ON) {
564 1 : CHK_RET(DeleteSocketWhiteList(hostPort, whitelist));
565 : }
566 14 : if (isByMasterInfo_ || !GetExternalInputHostPortSwitch()) {
567 14 : CHK_RET(listenSocket_->DeInit());
568 : } else {
569 0 : s32 deviceLogicId = INVALID_INT;
570 0 : CHK_RET(hrtGetDevice(&deviceLogicId));
571 0 : CHK_RET(PreemptPortManager::GetInstance(deviceLogicId).Release(listenSocket_));
572 : }
573 14 : listenSocket_ = nullptr;
574 : }
575 16 : return HCCL_SUCCESS;
576 : }
577 :
578 15 : HcclResult TopoInfoExchangeServer::StopNetwork(const std::vector<HcclIpAddress> &whitelist,
579 : u32 hostPort)
580 : {
581 15 : std::unique_lock<std::mutex> lock(lock_);
582 15 : CHK_RET(StopSocketListen(whitelist, hostPort));
583 :
584 15 : netDevCtx_ = nullptr;
585 15 : return HCCL_SUCCESS;
586 15 : }
587 :
588 0 : HcclResult TopoInfoExchangeServer::RecvRemoteAgentID(std::shared_ptr<HcclSocket> socket, std::string& agentID)
589 : {
590 0 : char agentBuf[MAX_AGENT_BUF_SIZE] = {0};
591 0 : HcclResult ret = socket->Recv(agentBuf, sizeof(agentBuf));
592 0 : agentBuf[MAX_AGENT_BUF_SIZE - 1] = '\0';
593 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
594 : HCCL_ERROR("[Recv][RemoteRankID]GetRemoteRankID receive rank id failed. ret[%d] ", ret), ret);
595 0 : agentID = agentBuf;
596 0 : return HCCL_SUCCESS;
597 : }
598 :
599 0 : HcclResult TopoInfoExchangeServer::RecvRemoteRankNum(std::shared_ptr<HcclSocket> socket, u32& remoteRankNum)
600 : {
601 0 : HcclResult ret = socket->Recv(reinterpret_cast<char *>(&remoteRankNum), sizeof(remoteRankNum));
602 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
603 : HCCL_ERROR("[Recv][RemoteRankNum]GetRemoteRankID receive rank num failed. ret[%d]", ret), ret);
604 0 : CHK_PRT_RET((remoteRankNum == 0), HCCL_ERROR("[Recv][RemoteRankNum]GetRemoteRankNum receive rank num "\
605 : "failed. rank num is zero."), HCCL_E_INTERNAL);
606 0 : return HCCL_SUCCESS;
607 : }
608 :
609 0 : HcclResult TopoInfoExchangeServer::VerifyRemoteRankNum(u32& previousRankNum, u32 remoteRankNum) const
610 : {
611 0 : if (previousRankNum == 0) {
612 0 : previousRankNum = remoteRankNum;
613 : } else {
614 0 : CHK_PRT_RET((remoteRankNum != previousRankNum),
615 : HCCL_ERROR("[Verify][RemoteRankNum]VerifyRemoteRankNum failed. remoteRankNum[%u] is difference "\
616 : "with others[%u].", remoteRankNum, previousRankNum), HCCL_E_INTERNAL);
617 : }
618 0 : return HCCL_SUCCESS;
619 : }
620 :
621 13 : HcclResult TopoInfoExchangeServer::GetRanksBasicInfo(
622 : const std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, RankTable_t &rankTable)
623 : {
624 : HcclResult ret;
625 13 : u32 socketIndex = 0; // socket已经经过rankid（or superPodId + serverip + deviceid排序）
626 13 : for (auto &handle : connectSockets) {
627 0 : ret = GetRankBasicInfo(handle.second, rankTable);
628 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
629 : HCCL_ERROR("[Get][RanksBasicInfo]GetRankBasicInfo from agentId[%s] failed, ret[%d]",
630 : handle.first.c_str(), ret), ret);
631 0 : if (isByMasterInfo_ && rankTable.rankList.size() > 0) { // masterInfo场景下无法获取rankid
632 0 : rankTable.rankList.back().rankId = socketIndex;
633 0 : connectSocketsWithRankID_.insert({socketIndex, handle.second});
634 : }
635 :
636 0 : HCCL_INFO("GetRankBasicInfo from agentId[%s] rankId[%u] success.",
637 : handle.first.c_str(), rankTable.rankList.back().rankId);
638 0 : socketIndex ++;
639 : }
640 13 : CHK_RET(SortRankList(rankTable));
641 13 : currentStep_++;
642 13 : return HCCL_SUCCESS;
643 : }
644 :
645 0 : HcclResult TopoInfoExchangeServer::GetRanksTransInfo(
646 : const std::map<std::string, std::shared_ptr<HcclSocket>> &connectSockets, RankTable_t &rankTable)
647 : {
648 : HcclResult ret;
649 0 : u32 socketIndex = 0;
650 0 : for (auto &handle : connectSockets) {
651 0 : RankTable_t tmpRankTable;
652 0 : ret = RecvClusterInfoMsg(handle.second, tmpRankTable);
653 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
654 : HCCL_ERROR("[Get][RanksTransInfo]RecvClusterInfoMsg from rank[%s] failed, ret[%u]", handle.first.c_str(),
655 : ret),
656 : ret);
657 0 : CHK_PRT_RET(tmpRankTable.rankList.size() == 0,
658 : HCCL_ERROR("[Get][RanksTransInfo]received rank list "
659 : "is empty."),
660 : HCCL_E_INTERNAL);
661 0 : for (u32 i = 0; i < tmpRankTable.rankList.size(); i++) {
662 0 : u32 currRank = isByMasterInfo_ ? socketIndex : tmpRankTable.rankList[i].rankId;
663 0 : if ((tmpRankTable.rankList[i].transportInfo.size()) != 0) {
664 0 : if (rankTable.rankList[currRank].transportInfo.size() == 0) {
665 0 : rankTable.rankList[currRank] = tmpRankTable.rankList[i];
666 : } else {
667 0 : HCCL_ERROR("[Get][RanksTransInfo]GetRanksTransInfo: rank[%u] transportInfo has existed.", currRank);
668 0 : return HCCL_E_INTERNAL;
669 : }
670 : }
671 : }
672 0 : socketIndex++;
673 0 : HCCL_INFO("RecvClusterInfoMsg from rank[%s] success.", handle.first.c_str());
674 0 : }
675 0 : currentStep_++;
676 0 : return HCCL_SUCCESS;
677 : }
678 :
679 0 : HcclResult TopoInfoExchangeServer::SendIdentify(std::shared_ptr<HcclSocket> socket, u32 identify) const
680 : {
681 0 : HcclResult ret = socket->Send(&identify, sizeof(identify));
682 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
683 : HCCL_ERROR("[Send][ClusterInfoMsg]errNo[0x%016llx] ra send identify failed! "\
684 : "ret[%u]", HCCL_ERROR_CODE(HCCL_E_TCP_TRANSFER), ret), ret);
685 :
686 0 : return HCCL_SUCCESS;
687 : }
688 :
689 0 : HcclResult TopoInfoExchangeServer::GetRankBasicInfo(std::shared_ptr<HcclSocket> socket, RankTable_t &rankTable)
690 : {
691 0 : RankTable_t tmpRankTable;
692 0 : CHK_RET(RecvClusterInfoMsg(socket, tmpRankTable));
693 :
694 0 : CHK_PRT_RET(tmpRankTable.rankList.size() == 0, HCCL_ERROR("[Get][RankBasicInfo]received rank list is "\
695 : "empty."), HCCL_E_INTERNAL);
696 0 : CHK_PRT_RET(tmpRankTable.serverList.size() == 0, HCCL_ERROR("[Get][RankBasicInfo]received server list "\
697 : "is empty."), HCCL_E_INTERNAL);
698 :
699 0 : for (u32 i = 0; i < tmpRankTable.rankList.size(); i++) {
700 0 : rankTable.rankList.push_back(tmpRankTable.rankList[i]);
701 : }
702 :
703 0 : if (rankTable.serverList.size() == 0) {
704 0 : rankTable.serverList = tmpRankTable.serverList;
705 : } else {
706 0 : for (u32 i = 0; i < tmpRankTable.serverList.size(); i++) {
707 0 : if (!DoServerIdExist(rankTable, tmpRankTable.serverList[i].serverId)) {
708 0 : rankTable.serverList.push_back(tmpRankTable.serverList[i]);
709 : }
710 : }
711 : }
712 :
713 0 : CHK_RET(GetCommonTopoInfo(rankTable, tmpRankTable));
714 :
715 0 : return HCCL_SUCCESS;
716 0 : }
717 :
718 0 : bool TopoInfoExchangeServer::DoServerIdExist(const RankTable_t& rankTable, const std::string& serverId) const
719 : {
720 0 : for (u32 i = 0; i < rankTable.serverList.size(); i++) {
721 0 : if (rankTable.serverList[i].serverId == serverId) {
722 0 : return true;
723 : }
724 : }
725 0 : return false;
726 : }
727 :
728 0 : HcclResult TopoInfoExchangeServer::GetCommonTopoInfo(RankTable_t &rankTable, const RankTable_t &orginRankTable) const
729 : {
730 0 : if (rankTable.rankNum == 0) {
731 0 : rankTable.nicDeploy = orginRankTable.nicDeploy;
732 0 : HCCL_INFO("get rank basicInfo nicDeploy[%u]", rankTable.nicDeploy);
733 : } else {
734 0 : CHK_PRT_RET(rankTable.nicDeploy != orginRankTable.nicDeploy,
735 : HCCL_ERROR("[Get][CommonTopoInfo]compare nicDeploy failed. curr[%u], recv[%u]",
736 : rankTable.nicDeploy, orginRankTable.nicDeploy), HCCL_E_INTERNAL);
737 : }
738 :
739 0 : rankTable.serverNum = rankTable.serverList.size();
740 0 : rankTable.rankNum = rankTable.rankList.size();
741 0 : CHK_RET(GetDevNum(rankTable.rankList, rankTable.deviceNum));
742 0 : CHK_RET(GetSuperPodNum(rankTable.rankList, rankTable.superPodNum));
743 0 : HCCL_INFO("get rank basicInfo serverNum[%u] rankNum[%u] deviceNum[%u] superPodNum[%u], nicDeploy[%u].",
744 : rankTable.serverNum, rankTable.rankNum, rankTable.deviceNum, rankTable.superPodNum, rankTable.nicDeploy);
745 0 : return HCCL_SUCCESS;
746 : }
747 :
748 0 : bool RankIdCompare(const RankInfo_t& i, const RankInfo_t& j)
749 : {
750 0 : return (i.rankId > j.rankId);
751 : }
752 :
753 13 : HcclResult TopoInfoExchangeServer::SortRankList(RankTable_t &rankTable) const
754 : {
755 13 : std::sort(rankTable.rankList.begin(), rankTable.rankList.end(), RankIdCompare);
756 13 : return HCCL_SUCCESS;
757 : }
758 : }
|
