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