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_remote_access.h"
12 : #include "externalinput_pub.h"
13 :
14 : namespace hccl {
15 : using namespace std;
16 :
17 0 : CommRemoteAccess::CommRemoteAccess(
18 : u32 rank, u32 devicePhyId, const std::map<u32, std::vector<HcclIpAddress>>& rankInfo,
19 0 : const std::vector<MemRegisterAddr>& addrInfos)
20 0 : : remoteTransportMap_(),
21 0 : rank_(rank),
22 0 : deviceLogicId_(0),
23 0 : devicePhyId_(devicePhyId),
24 0 : rankSize_(0),
25 0 : nicDeployment_(NICDeployment::NIC_DEPLOYMENT_DEVICE),
26 0 : rankInfo_(rankInfo),
27 0 : addrInfos_(addrInfos),
28 0 : dstInterServerMap_(),
29 0 : dstInterClientMap_(),
30 0 : nicSocketHandle_(),
31 0 : tag_("RemoteAccess"),
32 0 : threadsApplyNum_(0),
33 0 : dispatcher_(nullptr),
34 0 : notifyPool_(nullptr)
35 0 : {}
36 :
37 0 : CommRemoteAccess::~CommRemoteAccess()
38 : {
39 : HcclResult ret;
40 0 : for (u32 index = 0; index < linkThreads_.size(); index++) {
41 0 : if (linkThreads_[index]) {
42 0 : if (linkThreads_[index]->joinable()) {
43 0 : HCCL_DEBUG("Joining Link Thread[%u]", index);
44 0 : linkThreads_[index]->join(); // 等待线程执行后释放资源
45 : }
46 0 : ret = hrtResetDevice(deviceLogicId_); // 防止线程里面异常退出，在进程中reset
47 0 : if (ret != HCCL_SUCCESS) {
48 0 : HCCL_ERROR("[Comm][RemoteAccess]CommRemoteAccess reset device[%d] failed", deviceLogicId_);
49 : }
50 : }
51 : }
52 0 : struct SocketCloseInfoT socketCloseInfo = {};
53 0 : for (u32 i = 0; i < raSockets_.size(); i++) {
54 0 : socketCloseInfo.socketHandle = raSockets_[i].socketHandle; // 带入设备ID为物理ID
55 0 : socketCloseInfo.fdHandle = raSockets_[i].fdHandle;
56 0 : if ((raSockets_[i].socketHandle != nullptr) && (raSockets_[i].fdHandle != nullptr)) {
57 0 : ret = hrtRaSocketBatchClose(&socketCloseInfo, 1); /* 销毁已经创建的socket */
58 0 : if (ret != HCCL_SUCCESS) {
59 0 : HCCL_WARNING("~CommRemoteAccess:socket batch close fail! ret=%d", ret);
60 : }
61 : }
62 : }
63 :
64 0 : ret = DeleteSocketWhiteList();
65 0 : if (ret != HCCL_SUCCESS) {
66 0 : HCCL_WARNING("~CommRemoteAccess:delete Socket whiteList fail! ret=%d", ret);
67 : }
68 0 : remoteTransportMap_.clear();
69 :
70 0 : ret = CommRemoteDeInitRa();
71 0 : if (ret != HCCL_SUCCESS) {
72 0 : HCCL_ERROR("[Comm][RemoteAccess]CommRemoteAccess CommRemoteDeInitRa fail! ret[%d]", ret);
73 : }
74 0 : ret = notifyPool_->UnregisterOp(tag_);
75 0 : if (ret != HCCL_SUCCESS) {
76 0 : HCCL_WARNING("~CommRemoteAccess:UnregisterOp fail! ret=%d", ret);
77 : }
78 :
79 0 : if (dispatcher_ != nullptr) {
80 0 : HcclDispatcherDestroy(dispatcher_);
81 0 : dispatcher_ = nullptr;
82 : }
83 0 : }
84 :
85 0 : HcclResult CommRemoteAccess::Init()
86 : {
87 0 : HCCL_INFO("CommRemoteAccess Init start");
88 : // 获取当前线程操作的设备ID
89 0 : CHK_RET(hrtGetDevice(&deviceLogicId_));
90 : // dispatcher 资源初始化
91 0 : CHK_RET(RescoucePrepare());
92 : // 初始化ra资源，若hcom_init则不会进行再初始化
93 0 : CHK_RET(CommRemoteInitRa());
94 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo_));
95 : // 建链关系计算，只和统一平面和本端rank建单向链
96 0 : CHK_RET(CalcRemoteLink());
97 : // socket资源准备（白名单、batch connect
98 0 : CHK_RET(PrepareSocket());
99 : // 创建链接并保存fd_socket_handle
100 0 : CHK_RET(CreateLinks());
101 0 : HCCL_INFO("CommRemoteAccess Init end");
102 0 : return HCCL_SUCCESS;
103 : }
104 :
105 0 : HcclResult CommRemoteAccess::DeleteSocketWhiteList()
106 : {
107 0 : if (wlistInfosVec_.size() > 0) {
108 0 : for (u32 idx = 0; idx < nicSocketHandle_.size(); idx++) {
109 0 : CHK_RET(hrtRaSocketWhiteListDel(nicSocketHandle_[idx], wlistInfosVec_.data(), wlistInfosVec_.size()));
110 : }
111 : }
112 0 : return HCCL_SUCCESS;
113 : }
114 :
115 0 : std::shared_ptr<TransportRemoteAccess>& CommRemoteAccess::GetTransportByRank(const u32 dstRank)
116 : {
117 0 : if (remoteTransportMap_.find(dstRank) == remoteTransportMap_.end()) {
118 0 : HCCL_ERROR(
119 : "[Get][TransportByRank]can not find dstRank[%u] in remoteTransportMap_,"
120 : "remoteTransportMap_ size is [%llu]",
121 : dstRank, remoteTransportMap_.size());
122 0 : return transportDummy_;
123 : }
124 :
125 0 : return remoteTransportMap_.lower_bound(dstRank)->second;
126 : }
127 :
128 0 : HcclResult CommRemoteAccess::RescoucePrepare()
129 : {
130 : // 根据设备ID创建dispatcher
131 0 : CHK_SMART_PTR_NULL(dispatcher_);
132 :
133 0 : notifyPool_.reset(new (std::nothrow) NotifyPool());
134 0 : CHK_SMART_PTR_NULL(notifyPool_);
135 0 : CHK_RET(notifyPool_->Init(devicePhyId_));
136 0 : CHK_RET(notifyPool_->RegisterOp(tag_));
137 0 : return HCCL_SUCCESS;
138 : }
139 :
140 0 : HcclResult CommRemoteAccess::CommRemoteInitRa()
141 : {
142 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).Init(nicDeployment_));
143 :
144 0 : auto iter = rankInfo_.find(rank_);
145 0 : bool check = (iter == rankInfo_.end());
146 0 : CHK_PRT_RET(check, HCCL_ERROR("[Get][Instance]can not find rank[%u] info in rankInfo_", rank_), HCCL_E_PARA);
147 0 : HCCL_INFO("in CommRemoteInitRa, rank_[%u], iter->second.size[%d]", rank_, iter->second.size());
148 0 : for (size_t ipIdex = 0; ipIdex < iter->second.size(); ipIdex++) {
149 0 : if (iter->second[ipIdex].IsInvalid()) {
150 0 : continue;
151 : }
152 0 : u32 port = HETEROG_CCL_PORT;
153 0 : HcclResult ret = NetworkManager::GetInstance(deviceLogicId_).StartNic(iter->second[ipIdex], port, true);
154 0 : CHK_PRT_RET(
155 : ret != HCCL_SUCCESS,
156 : HCCL_ERROR("[InitRa][CommRemote]start nic ipaddr[%s] failed", iter->second[ipIdex].GetReadableAddress()),
157 : ret);
158 : }
159 0 : return HCCL_SUCCESS;
160 : }
161 :
162 0 : HcclResult CommRemoteAccess::CommRemoteDeInitRa()
163 : {
164 0 : auto iter = rankInfo_.find(rank_);
165 0 : bool check = (iter == rankInfo_.end());
166 0 : CHK_PRT_RET(check, HCCL_ERROR("[DeInit][Ra]can not find rank[%u] info in rankInfo_", rank_), HCCL_E_PARA);
167 0 : HCCL_INFO("in CommRemoteDeInitRa, rank_[%u], iter->second.size[%zu]", rank_, iter->second.size());
168 0 : for (size_t ipIdex = 0; ipIdex < iter->second.size(); ipIdex++) {
169 0 : if (iter->second[ipIdex].IsInvalid()) {
170 0 : continue;
171 : }
172 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).StopNic(iter->second[ipIdex], 0));
173 : }
174 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).DeInit(nicDeployment_));
175 0 : return HCCL_SUCCESS;
176 : }
177 :
178 0 : HcclResult CommRemoteAccess::CalcRemoteLink()
179 : {
180 0 : rankSize_ = rankInfo_.size();
181 0 : if ((rankSize_ == 0)) {
182 0 : HCCL_ERROR("[Calc][RemoteLink]invalid rankSize, rankSize:[%zu].", rankSize_);
183 0 : return HCCL_E_PARA;
184 : }
185 :
186 : // 计算同一平面的server端rank信息，小于或等于本rank的为server端（本rank既是server也是client）
187 0 : for (u32 serverRank = rank_;; serverRank--) {
188 0 : HCCL_INFO("CalcRemoteLink rank[%u] serverRank[%u]", rank_, serverRank);
189 0 : auto serverRankInfo = rankInfo_.find(serverRank);
190 0 : if (serverRankInfo == rankInfo_.end()) {
191 0 : HCCL_ERROR("[Calc][RemoteLink]can not find server rank[%u] in rankInfo_.", serverRank);
192 0 : return HCCL_E_PARA;
193 : }
194 0 : std::vector<HcclIpAddress> ipVec(rankInfo_[serverRank]);
195 0 : auto serverIter = dstInterServerMap_.find(serverRank);
196 0 : if (serverIter == dstInterServerMap_.end()) {
197 0 : dstInterServerMap_.insert({serverRank, ipVec});
198 : }
199 0 : if (serverRank == 0) {
200 0 : break;
201 : }
202 0 : }
203 : // 计算同一平面的client端rank信息，大于等于本rank的为client端（本rank既是server也是client）
204 0 : for (u32 clientRank = rank_; clientRank < rankSize_; clientRank++) {
205 0 : auto clientRankInfo = rankInfo_.find(clientRank);
206 0 : if (clientRankInfo == rankInfo_.end()) {
207 0 : HCCL_ERROR("[Calc][RemoteLink]can not find client rank[%u] in rankInfo_.", clientRank);
208 0 : return HCCL_E_PARA;
209 : }
210 0 : std::vector<HcclIpAddress> ipVec(rankInfo_[clientRank]);
211 0 : auto clientIter = dstInterClientMap_.find(clientRank);
212 0 : if (clientIter == dstInterClientMap_.end()) {
213 0 : dstInterClientMap_.insert({clientRank, ipVec});
214 : }
215 0 : }
216 0 : return HCCL_SUCCESS;
217 : }
218 :
219 0 : HcclResult CommRemoteAccess::PrepareSocket()
220 : {
221 : // socket handle 是一一对应关系
222 0 : if (dstInterServerMap_.size() * raResourceInfo_.nicSocketMap.size()
223 0 : + dstInterClientMap_.size() * raResourceInfo_.nicSocketMap.size()
224 0 : > 0) {
225 0 : for (u32 idx = 0; idx < rankInfo_[rank_].size(); idx++) {
226 0 : if (rankInfo_[rank_][idx].IsInvalid()) {
227 0 : HCCL_ERROR("[Prepare][Socket]rank_[%u] nicIp[%u] is 0", rank_, idx);
228 0 : continue;
229 : }
230 0 : auto it = raResourceInfo_.nicSocketMap.find(rankInfo_[rank_][idx]);
231 0 : if (it == raResourceInfo_.nicSocketMap.end()) {
232 0 : HCCL_ERROR(
233 : "[Prepare][Socket]can not find nicSocketHandle, ip[%s]",
234 : rankInfo_[rank_][idx].GetReadableAddress());
235 0 : return HCCL_E_PARA;
236 : } else {
237 0 : if (it->second.nicSocketHandle == nullptr) {
238 0 : HCCL_ERROR(
239 : "[Prepare][Socket]CommRemoteAccess prepare socket failed! rank[%u] IP addr[%s]", rank_,
240 : rankInfo_[rank_][idx].GetReadableAddress());
241 0 : return HCCL_E_PARA;
242 : }
243 0 : nicSocketHandle_.push_back(it->second.nicSocketHandle);
244 : }
245 0 : HCCL_INFO(
246 : "rank[%u], nicSocketMap[%u] nicIp.size[%u]", rank_, raResourceInfo_.nicSocketMap.size(),
247 : rankInfo_[rank_].size());
248 : }
249 : }
250 :
251 0 : HCCL_INFO("In PrepareSocket raResourceInfo_Size[%u]", nicSocketHandle_.size());
252 0 : CHK_RET(AddSocketWhiteList());
253 :
254 : // 当前rank作为client端batch connect动作
255 0 : u32 dstInterServerNum = dstInterServerMap_.size() * raResourceInfo_.nicSocketMap.size();
256 0 : HCCL_INFO("socket batch connect dstInterServerNum[%u]", dstInterServerNum);
257 0 : if (dstInterServerNum > 0) {
258 0 : std::vector<struct SocketConnectInfoT> conns(dstInterServerNum);
259 0 : struct SocketConnectInfoT* conn = conns.data();
260 0 : s32 sRet = memset_s(
261 0 : conn, sizeof(struct SocketConnectInfoT) * dstInterServerNum, 0,
262 0 : sizeof(struct SocketConnectInfoT) * dstInterServerNum);
263 0 : CHK_PRT_RET(
264 : sRet != EOK,
265 : HCCL_ERROR(
266 : "[Prepare][Socket]memory set failed, return[%d]. params:"
267 : "destMaxSize[%zu], c[%d], count[%zu]",
268 : sRet, sizeof(struct SocketConnectInfoT) * dstInterServerNum, 0,
269 : sizeof(struct SocketConnectInfoT) * dstInterServerNum),
270 : HCCL_E_MEMORY);
271 :
272 0 : u32 loop = 0;
273 0 : for (auto iter = dstInterServerMap_.begin(); iter != dstInterServerMap_.end(); iter++) {
274 0 : for (u32 idx = 0; idx < nicSocketHandle_.size() && idx < (iter->second).size(); idx++) {
275 0 : conn[loop].remoteIp.addr = (iter->second)[idx].GetBinaryAddress().addr;
276 0 : conn[loop].remoteIp.addr6 = (iter->second)[idx].GetBinaryAddress().addr6;
277 0 : conn[loop].socketHandle = nicSocketHandle_[idx];
278 0 : conn[loop].port = HETEROG_CCL_PORT;
279 0 : if (nicSocketHandle_[idx] == nullptr) {
280 0 : HCCL_ERROR("[Prepare][Socket]index[%u] nicSocketHandle_ is null", idx);
281 0 : return HCCL_E_INTERNAL;
282 : }
283 0 : sRet = memcpy_s(&conn[loop].tag[0], sizeof(conn[loop].tag) - 1, tag_.c_str(), tag_.size());
284 0 : CHK_PRT_RET(
285 : sRet != EOK,
286 : HCCL_ERROR(
287 : "[Prepare][Socket]memcpy failed. errorno[%d]:"
288 : "destMaxSize[%zu], count[%zu]",
289 : sRet, sizeof(conn[loop].tag), tag_.size()),
290 : HCCL_E_MEMORY);
291 : }
292 0 : loop++;
293 : }
294 0 : HcclResult ret = hrtRaSocketBatchConnect(conn, dstInterServerNum);
295 0 : CHK_PRT_RET(
296 : ret != HCCL_SUCCESS,
297 : HCCL_ERROR(
298 : "[Prepare][Socket]socket batch failed, batch size[%u], loop[%u], "
299 : "dst_inter_server_map_size[%u], handle size[%u]",
300 : dstInterServerNum, loop, dstInterServerMap_.size(), nicSocketHandle_.size()),
301 : ret);
302 0 : }
303 :
304 0 : return HCCL_SUCCESS;
305 : }
306 :
307 0 : HcclResult CommRemoteAccess::AddSocketWhiteList()
308 : {
309 : // 当前rank作为server端socket白名单下发动作
310 0 : struct SocketWlistInfoT wlistInfo = {};
311 0 : for (auto iter = dstInterClientMap_.begin(); iter != dstInterClientMap_.end(); iter++) {
312 0 : wlistInfo.connLimit = NIC_SOCKET_CONN_LIMIT;
313 0 : s32 sRet = memcpy_s(&wlistInfo.tag[0], sizeof(wlistInfo.tag) - 1, tag_.c_str(), tag_.size());
314 0 : if (sRet != EOK) {
315 0 : HCCL_ERROR(
316 : "[Add][Socket]memory copy failed. errorno[%d]: dest size[%zu], src[%s],"
317 : "count[%zu]",
318 : sRet, sizeof(wlistInfo.tag), tag_.c_str(), tag_.size());
319 0 : wlistInfosVec_.clear();
320 0 : return HCCL_E_MEMORY;
321 : }
322 0 : for (u32 idx = 0; idx < (iter->second).size(); idx++) {
323 0 : wlistInfo.remoteIp.addr = iter->second[idx].GetBinaryAddress().addr;
324 0 : wlistInfo.remoteIp.addr6 = iter->second[idx].GetBinaryAddress().addr6;
325 0 : wlistInfosVec_.push_back(wlistInfo);
326 : }
327 : }
328 :
329 0 : if (wlistInfosVec_.size() > 0) {
330 0 : for (u32 idx = 0; idx < nicSocketHandle_.size(); idx++) {
331 0 : CHK_RET(hrtRaSocketWhiteListAdd(nicSocketHandle_[idx], wlistInfosVec_.data(), wlistInfosVec_.size()));
332 : }
333 : }
334 :
335 0 : return HCCL_SUCCESS;
336 : }
337 :
338 0 : HcclResult CommRemoteAccess::CreateLinks()
339 : {
340 : // 计算建链所需线程
341 0 : u32 nicNum = raResourceInfo_.nicSocketMap.size();
342 0 : u32 threadsNum = dstInterClientMap_.size() * nicNum + dstInterServerMap_.size() * nicNum;
343 0 : HCCL_INFO("threadsNum[%u]", threadsNum);
344 0 : HCCL_INFO(
345 : "CommRemoteAccess CreateLinks rank_[%u] dstInterClientMapSize[%u], nicNum[%u], dstInterServerMapSize[%u]",
346 : rank_, dstInterClientMap_.size(), nicNum, dstInterServerMap_.size());
347 0 : CHK_PRT_RET(
348 : (threadsNum == 0),
349 : HCCL_ERROR(
350 : "[Create][Links]no link to create, please check ranktable to see if"
351 : "device_ip is configured. nicNum[%u]",
352 : nicNum),
353 : HCCL_E_INTERNAL);
354 0 : linkThreads_.resize(threadsNum);
355 0 : threadsStatus_.resize(threadsNum);
356 0 : HCCL_INFO(
357 : "comm base threads info:link threads size[%llu], dst inter client map size[%llu], "
358 : "dst inter server map size[%llu]",
359 : linkThreads_.size(), dstInterClientMap_.size() * nicNum, dstInterServerMap_.size() * nicNum);
360 0 : HcclUs startut = TIME_NOW();
361 : // 获取当前rank作为client端时，获取所有server端的socket
362 0 : CHK_RET(CreateInterServerLinks());
363 : // 获取当前rank作为server端时，获取所有client端的socket
364 0 : CHK_RET(CreateInterClientLinks());
365 :
366 0 : bool check = (threadsApplyNum_ != linkThreads_.size());
367 0 : CHK_PRT_RET(
368 : check,
369 : HCCL_ERROR(
370 : "[Create][Links]comm apply num[%u] is not equal to link threads[%llu]", threadsApplyNum_,
371 : linkThreads_.size()),
372 : HCCL_E_INTERNAL);
373 :
374 0 : HCCL_INFO("CommRemoteAccess CreateLinks threadsApplyNum_[%u]", threadsApplyNum_);
375 :
376 0 : for (u32 index = 0; index < linkThreads_.size(); index++) {
377 0 : linkThreads_[index]->join(); // 等待线程执行完毕
378 0 : CHK_RET(hrtResetDevice(deviceLogicId_)); // 防止线程里面异常退出，在进程中reset
379 : }
380 0 : for (u32 index = 0; index < threadsStatus_.size(); index++) {
381 0 : CHK_PRT_RET(
382 : threadsStatus_[index] != 0, HCCL_ERROR("[Create][Links]execute the thread[%u] function failed", index),
383 : HCCL_E_PARA);
384 : }
385 0 : linkThreads_.clear();
386 0 : HCCL_DEBUG("rdma_rasocket Time:%lld us", DURATION_US(TIME_NOW() - startut));
387 0 : return HCCL_SUCCESS;
388 : }
389 :
390 0 : HcclResult CommRemoteAccess::CreateInterServerLinks()
391 : {
392 : // 获取当前rank作为client端时，获取所有server端的socket
393 0 : u32 dstInterServerNum = dstInterServerMap_.size() * nicSocketHandle_.size();
394 0 : if (dstInterServerNum > 0) {
395 0 : std::vector<struct SocketInfoT> cliConns(dstInterServerNum);
396 0 : struct SocketInfoT* cliConn = cliConns.data();
397 0 : s32 sRet = memset_s(
398 0 : cliConn, sizeof(struct SocketInfoT) * dstInterServerNum, 0, sizeof(struct SocketInfoT) * dstInterServerNum);
399 0 : CHK_PRT_RET(
400 : sRet != EOK,
401 : HCCL_ERROR(
402 : "[Create][InterServerLinks]memory set failed. return[%d]."
403 : "params: destMaxSize[%zu], c[%d], count[%zu]",
404 : sRet, sizeof(struct SocketInfoT) * dstInterServerNum, 0,
405 : sizeof(struct SocketInfoT) * dstInterServerNum),
406 : HCCL_E_MEMORY);
407 : // 构建socket_info_t信息，用于获取fd_socket_handle
408 0 : u32 connLoop = 0;
409 0 : for (auto iter = dstInterServerMap_.begin(); iter != dstInterServerMap_.end(); iter++) {
410 0 : sRet = memcpy_s(&cliConn[connLoop].tag[0], sizeof(cliConn[connLoop].tag) - 1, tag_.c_str(), tag_.size());
411 0 : CHK_PRT_RET(
412 : sRet != EOK,
413 : HCCL_ERROR(
414 : "[Create][InterServerLinks]memcpy failed. errorno[%d],"
415 : "params:destMaxSize[%zu], count[%zu]",
416 : sRet, sizeof(cliConn[connLoop].tag), tag_.size()),
417 : HCCL_E_MEMORY);
418 0 : for (u32 idx = 0; idx < nicSocketHandle_.size(); idx++) {
419 0 : cliConn[connLoop].socketHandle = nicSocketHandle_[idx];
420 0 : cliConn[connLoop].remoteIp.addr = iter->second[idx].GetBinaryAddress().addr;
421 0 : cliConn[connLoop].remoteIp.addr6 = iter->second[idx].GetBinaryAddress().addr6;
422 : /* 插入建链状态的指示 */
423 0 : LinkStatus_t linkInfo;
424 0 : linkInfo.userRank = iter->first;
425 0 : linkInfo.status = SOCKET_CONNECT_NO_CONNECTION;
426 0 : linkInfo.isLinked = false;
427 0 : linkInfo.remoteIp = (iter->second)[idx];
428 0 : linkInfo.localIp = rankInfo_[rank_][idx];
429 0 : HCCL_DEBUG(
430 : "CLIENT rank[%u] LocalIp[%s] RemoteIp[%s]", rank_, linkInfo.localIp.GetReadableAddress(),
431 : linkInfo.remoteIp.GetReadableAddress());
432 0 : serverLinkStatus_.insert(std::make_pair(iter->second[idx], linkInfo));
433 0 : connLoop++;
434 0 : }
435 : }
436 0 : CHK_RET(GetRaSocket(CLIENT_ROLE_SOCKET, cliConn, dstInterServerNum));
437 0 : }
438 0 : return HCCL_SUCCESS;
439 : }
440 :
441 0 : HcclResult CommRemoteAccess::GetRaSocket(const u32 role, const struct SocketInfoT conn[], const u32 num)
442 : {
443 0 : HCCL_INFO("get sockets para: socket role[%u], socket num[%u]", role, num);
444 0 : auto startTime = std::chrono::steady_clock::now();
445 0 : auto timeout = std::chrono::seconds(GetExternalInputHcclLinkTimeOut());
446 : s32 sockRet;
447 0 : u32 gotSocketsCnt = 0;
448 0 : HCCL_INFO("In GetRaSocket, waiting for all rasockets link up...");
449 0 : u32 left = num;
450 : while (true) {
451 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
452 0 : PrintErrorConnection(role, left);
453 0 : HCCL_ERROR(
454 : "[Get][RaSocket]in GetRaSocket, get rasocket error role[%u], rank[%u]num[%u], timeout[%lld s]", role,
455 : rank_, left, timeout);
456 0 : return HCCL_E_TIMEOUT;
457 : }
458 0 : std::vector<struct SocketInfoT> conns(left);
459 0 : struct SocketInfoT* tmpConn = conns.data();
460 0 : s32 sret = memcpy_s(
461 0 : tmpConn, sizeof(struct SocketInfoT) * left, conn + (num - left), sizeof(struct SocketInfoT) * left);
462 0 : CHK_PRT_RET(
463 : sret != EOK,
464 : HCCL_ERROR(
465 : "[Get][RaSocket]memcpy failed. errorno[%d], params:"
466 : "destMaxSize[%zu], count[%zu]",
467 : sret, sizeof(struct SocketInfoT) * left, sizeof(struct SocketInfoT) * left),
468 : HCCL_E_MEMORY);
469 0 : u32 connectedNum = 0;
470 0 : sockRet = hrtRaGetSockets(role, tmpConn, left, &connectedNum);
471 0 : if ((connectedNum == 0 && sockRet == 0) || (sockRet == SOCK_EAGAIN)) {
472 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
473 0 : } else if (sockRet != 0) {
474 0 : PrintErrorConnection(role, num);
475 0 : HCCL_ERROR(
476 : "[Get][RaSocket]in GetRaSocket, get rasocket error. role[%u], rank[%u],num[%u] sockRet[%d] > 0", role,
477 : rank_, num, sockRet);
478 0 : return HCCL_E_TCP_CONNECT;
479 0 : } else if (connectedNum > 0) {
480 0 : u32 sockNum = abs(static_cast<s32>(connectedNum));
481 0 : left = left - sockNum;
482 : // 保存建链成功的socket
483 0 : HcclResult ret = DealSuccRasocket(connectedNum, role, tmpConn, sockNum);
484 0 : CHK_PRT_RET(
485 : ret != HCCL_SUCCESS,
486 : HCCL_ERROR(
487 : "[Get][RaSocket]in GetRaSocket, save rasocket failed. role[%u], rank[%u]"
488 : "num[%u] ret[%d] connectednum[%u]",
489 : role, rank_, num, ret, connectedNum),
490 : ret);
491 0 : gotSocketsCnt += sockNum;
492 :
493 0 : if (gotSocketsCnt == num) {
494 0 : break;
495 0 : } else if (gotSocketsCnt > num) {
496 0 : HCCL_ERROR("[Get][RaSocket]total Sockets[%u], more than needed num[%u]!", gotSocketsCnt, num);
497 0 : return HCCL_E_TCP_CONNECT;
498 : } else {
499 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
500 : }
501 : }
502 0 : }
503 0 : HCCL_INFO("In CommRemoteAccess, all rasockets linked up ");
504 0 : return HCCL_SUCCESS;
505 : }
506 :
507 0 : HcclResult CommRemoteAccess::CreateInterClientLinks()
508 : {
509 : // 获取当前rank作为server端时，获取所有client端的socket
510 0 : u32 dstInterClientNum = dstInterClientMap_.size() * nicSocketHandle_.size();
511 0 : HCCL_INFO("dstInterClientNum[%u]", dstInterClientNum);
512 0 : if (dstInterClientNum > 0) {
513 0 : std::vector<struct SocketInfoT> srvConns(dstInterClientNum);
514 0 : struct SocketInfoT* srvConn = srvConns.data();
515 0 : s32 sRet = memset_s(
516 0 : srvConn, sizeof(struct SocketInfoT) * dstInterClientNum, 0, sizeof(struct SocketInfoT) * dstInterClientNum);
517 0 : CHK_PRT_RET(
518 : sRet != EOK,
519 : HCCL_ERROR(
520 : "[Create][InterClientLinks]memory set failed. return[%d]."
521 : "params: destMaxSize[%zu], c[%d], count[%zu]",
522 : sRet, sizeof(struct SocketInfoT) * dstInterClientNum, 0,
523 : sizeof(struct SocketInfoT) * dstInterClientNum),
524 : HCCL_E_MEMORY);
525 0 : u32 loop = 0;
526 0 : for (u32 interIndex = 0; interIndex < dstInterClientMap_.size(); interIndex++) {
527 0 : sRet = memcpy_s(&srvConn[loop].tag[0], sizeof(srvConn[loop].tag) - 1, tag_.c_str(), tag_.size());
528 0 : CHK_PRT_RET(
529 : sRet != EOK,
530 : HCCL_ERROR(
531 : "[Create][InterClientLinks]memcpy failed. errorno[%d], params:"
532 : "destMaxSize[%zu],count[%zu]",
533 : sRet, sizeof(srvConn[loop].tag), tag_.size()),
534 : HCCL_E_MEMORY);
535 0 : for (u32 idx = 0; idx < nicSocketHandle_.size(); idx++) {
536 0 : srvConn[loop].socketHandle = nicSocketHandle_[idx];
537 0 : loop++;
538 : }
539 : }
540 : /* 插入建链状态的指示 */
541 0 : for (auto iter = dstInterClientMap_.begin(); iter != dstInterClientMap_.end(); iter++) {
542 0 : LinkStatus_t linkInfo;
543 0 : linkInfo.userRank = iter->first;
544 0 : linkInfo.status = SOCKET_CONNECT_NO_CONNECTION;
545 0 : linkInfo.isLinked = false;
546 0 : for (u32 idx = 0; idx < (iter->second).size(); idx++) {
547 0 : linkInfo.remoteIp = (iter->second)[idx];
548 0 : linkInfo.localIp = rankInfo_[rank_][idx];
549 0 : HCCL_DEBUG(
550 : "CreateInterClientLinks SERVER rank[%u] LocalIp[%s] RemoteIp[%s]", rank_,
551 : linkInfo.localIp.GetReadableAddress(), linkInfo.remoteIp.GetReadableAddress());
552 0 : clientLinkStatus_.insert(std::make_pair((iter->second)[idx], linkInfo));
553 : }
554 0 : }
555 0 : CHK_RET(GetRaSocket(SERVER_ROLE_SOCKET, srvConn, dstInterClientNum));
556 0 : }
557 0 : return HCCL_SUCCESS;
558 : }
559 :
560 0 : void CommRemoteAccess::PrintErrorConnection(const u32 role, const u32 num)
561 : {
562 0 : RPT_INNER_ERR_PRT("remote op nic connect failed, please ensure that collective communication execution status "
563 : "of each device is consistent(include network TLS configuration)");
564 :
565 0 : HCCL_ERROR("Some NPUs get socket timeout, the details are as follows:");
566 0 : HCCL_ERROR(" _________________________LINK_ERROR_INFO___________________________");
567 0 : HCCL_ERROR(" | comm error, device[%d] num[%u] ", deviceLogicId_, num);
568 0 : HCCL_ERROR(" | dest_ip(user_rank) | dest_port | src_ip(user_rank) | src_port | MyRole "
569 : "| Status |");
570 0 : HCCL_ERROR(" |--------------------|--------------------|----------|------------|-----------------"
571 : "|-----------------|");
572 :
573 : /* 第一行打印deviceIds */
574 0 : HcclResult ret = HCCL_SUCCESS;
575 0 : if (role == SERVER_ROLE_SOCKET) {
576 0 : ret = PrintErrorConnectionInfo(clientLinkStatus_, role);
577 0 : } else if (role == CLIENT_ROLE_SOCKET) {
578 0 : ret = PrintErrorConnectionInfo(serverLinkStatus_, role);
579 : }
580 0 : if (ret != HCCL_SUCCESS) {
581 0 : HCCL_ERROR("[Print][ErrorConnection]PrintErrorConnectionInfo fail. ret[%d] role[%u]", ret, role);
582 0 : return;
583 : }
584 0 : HCCL_ERROR(" ___________________________________________________________________ ");
585 0 : HCCL_ERROR("the connection failure between this device and target device may be due to the following reasons:");
586 0 : HCCL_ERROR("1. the connection between this device and the target device is abnormal.");
587 0 : HCCL_ERROR("2. an exception occurred at the target devices.");
588 0 : HCCL_ERROR("3. the time difference between the execution of hcom on this device and the target device exceeds the "
589 : "timeout threshold, make sure this by keyword [Entry-].");
590 0 : HCCL_ERROR("4. the behavior of executing the calculation graph on this device and the target device is "
591 : "inconsistent. ");
592 0 : HCCL_ERROR("5. Now you can freely specify a port for listening and connecting. If an invalid port is chosen, "
593 : "it may result in failed listening and connection timeouts");
594 0 : return;
595 : }
596 :
597 : #define TRANSFORM_RASOCKET_STATUS(status, stringStatus) \
598 : do { \
599 : switch (status) { \
600 : default: \
601 : case SOCKET_CONNECT_NO_CONNECTION: \
602 : stringStatus = "no connect"; \
603 : break; \
604 : case SOCKET_CONNECT_OK: \
605 : stringStatus = "connected"; \
606 : break; \
607 : case SOCKET_CONNECT_TIMEOUT: \
608 : stringStatus = "connecting"; \
609 : break; \
610 : } \
611 : } while (0)
612 :
613 : HcclResult
614 0 : CommRemoteAccess::PrintErrorConnectionInfo(const std::map<HcclIpAddress, LinkStatus_t>& linkStatusMap, u32 role)
615 : {
616 0 : std::string sRole;
617 0 : switch (role) {
618 0 : case SERVER_ROLE_SOCKET:
619 0 : sRole = " server ";
620 0 : break;
621 0 : case CLIENT_ROLE_SOCKET:
622 0 : sRole = " client ";
623 0 : break;
624 0 : default:
625 0 : sRole = " NA ";
626 0 : break;
627 : }
628 0 : for (auto iter = linkStatusMap.begin(); iter != linkStatusMap.end(); iter++) {
629 0 : if (!iter->second.isLinked) {
630 0 : std::string connectStatus = "";
631 0 : TRANSFORM_RASOCKET_STATUS(iter->second.status, connectStatus);
632 0 : HCCL_ERROR(
633 : " | %s(%u) | %u | %s(%u) | %u | %s | %s | ", iter->second.remoteIp.GetReadableAddress(),
634 : iter->second.userRank, HETEROG_CCL_PORT, iter->second.localIp.GetReadableAddress(), rank_,
635 : HETEROG_CCL_PORT, sRole.c_str(), connectStatus.c_str());
636 0 : }
637 : }
638 0 : return HCCL_SUCCESS;
639 0 : }
640 :
641 : // 根据IP信息，获得RANK信息
642 0 : HcclResult CommRemoteAccess::GetDstRank(
643 : std::map<u32, std::vector<HcclIpAddress>>& dstMap, const HcclIpAddress& dstIp, u32& dstRank)
644 : {
645 0 : for (auto it = dstMap.begin(); it != dstMap.end(); it++) {
646 0 : for (u32 idx = 0; idx < it->second.size(); idx++) {
647 0 : if (it->second[idx] == dstIp) {
648 0 : dstRank = it->first;
649 0 : return HCCL_SUCCESS;
650 : }
651 : }
652 : }
653 :
654 0 : HCCL_ERROR("[Get][DstRank]can't find ip[%s] in dst map", dstIp.GetReadableAddress());
655 0 : return HCCL_E_NOT_FOUND;
656 : }
657 :
658 0 : HcclResult CommRemoteAccess::CreateInterThread(const u32 role, const SocketInfoT& socketInfo)
659 : {
660 : // 线程命名，CommRemoteTerL代表CommRemote Inter Link
661 0 : std::string threadStr = "RemoteThrd_" + std::to_string(threadsApplyNum_);
662 0 : HcclIpAddress nicIp;
663 0 : u32 dstRank = 0;
664 0 : threadsStatus_[threadsApplyNum_] = 1;
665 0 : CHK_RET(GetNicByHandle(socketInfo.socketHandle, nicIp));
666 : HcclInAddr temp;
667 0 : temp.addr = socketInfo.remoteIp.addr;
668 0 : temp.addr6 = socketInfo.remoteIp.addr6;
669 0 : HcclIpAddress remoteIP(rankInfo_[rank_][0].GetFamily(), temp);
670 0 : CHK_PRT_RET(remoteIP.IsInvalid(), HCCL_ERROR("ip is invalid."), HCCL_E_PARA);
671 0 : workflowMode_ = GetWorkflowMode();
672 0 : if (role == SERVER_ROLE_SOCKET) {
673 0 : CHK_RET(GetDstRank(dstInterClientMap_, remoteIP, dstRank));
674 0 : linkThreads_[threadsApplyNum_].reset(new (std::nothrow) std::thread(
675 0 : &CommRemoteAccess::InitDestTransport, this, hrtErrMGetErrorContext(), role, nicIp, dstRank, threadStr,
676 0 : socketInfo.fdHandle, &threadsStatus_[threadsApplyNum_]));
677 : }
678 :
679 0 : if (role == CLIENT_ROLE_SOCKET) {
680 0 : CHK_RET(GetDstRank(dstInterServerMap_, remoteIP, dstRank));
681 0 : linkThreads_[threadsApplyNum_].reset(new (std::nothrow) std::thread(
682 0 : &CommRemoteAccess::InitDestTransport, this, hrtErrMGetErrorContext(), role, nicIp, dstRank, threadStr,
683 0 : socketInfo.fdHandle, &threadsStatus_[threadsApplyNum_]));
684 : }
685 0 : bool check = !linkThreads_[threadsApplyNum_];
686 0 : CHK_PRT_RET(
687 : check, HCCL_ERROR("[Create][InterThread]link threads[%u] reset failed.", threadsApplyNum_), HCCL_E_INTERNAL);
688 0 : threadsApplyNum_++;
689 0 : return HCCL_SUCCESS;
690 0 : }
691 :
692 : HcclResult
693 0 : CommRemoteAccess::DealSuccRasocket(s32 sockRet, const u32 role, const struct SocketInfoT tmpConn[], const u32 num)
694 : {
695 0 : HCCL_DEBUG("CommRemoteAccess DealSuccRasocketNum[%u]", num);
696 0 : u32 socketsCnt = static_cast<u32>(sockRet);
697 0 : u32 loop = 0;
698 0 : for (u32 i = 0; i < num; i++) {
699 : HcclInAddr temp;
700 0 : temp.addr = tmpConn[i].remoteIp.addr;
701 0 : temp.addr6 = tmpConn[i].remoteIp.addr6;
702 0 : HcclIpAddress remoteIP(rankInfo_[rank_][0].GetFamily(), temp);
703 0 : CHK_PRT_RET(remoteIP.IsInvalid(), HCCL_ERROR("ip is invalid."), HCCL_E_PARA);
704 0 : if (tmpConn[i].status == SOCKET_CONNECT_OK) {
705 0 : raSockets_.push_back(tmpConn[i]);
706 0 : CHK_RET(CreateInterThread(role, tmpConn[i]));
707 : // 建链成功的在本地标志建链成功
708 0 : serverLinkStatus_[remoteIP].isLinked = true;
709 0 : loop++;
710 : }
711 0 : if (tmpConn[i].status != SOCKET_CONNECT_NO_CONNECTION) {
712 0 : clientLinkStatus_[remoteIP].status = tmpConn[i].status;
713 : }
714 0 : }
715 :
716 0 : if (socketsCnt != loop) {
717 0 : HCCL_ERROR(
718 : "[Deal][SuccRasocket]current socketsCnt[%u], not equal to actual connect number[%u]!", socketsCnt, loop);
719 0 : return HCCL_E_TCP_CONNECT;
720 : }
721 0 : return HCCL_SUCCESS;
722 : }
723 :
724 0 : HcclResult CommRemoteAccess::InitDestTransport(
725 : const ErrContext& error_context, u32 role, const HcclIpAddress& nicIp, const u32 dstRank,
726 : const std::string& threadStr, FdHandle socketFdHandle, u32* getThreadStatus)
727 : {
728 0 : hrtErrMSetErrorContext(error_context);
729 :
730 : // 给当前线程添加名字
731 0 : SetThreadName(threadStr);
732 0 : CHK_RET(hrtSetDevice(deviceLogicId_));
733 0 : SetWorkflowMode(workflowMode_);
734 :
735 0 : RemoteAccessPara accessPara;
736 0 : CHK_RET(SetAccessPara(role, nicIp, dstRank, socketFdHandle, accessPara));
737 0 : HCCL_INFO(
738 : "[InitDestTransport para]local_rank[%u]-localIpAddr[%s],dst rank[%u]-remote_rank[%u]-remote_ip_addr[%s], "
739 : "role[%u]",
740 : rank_, rankInfo_[rank_][0].GetReadableAddress(), dstRank, dstRank, rankInfo_[dstRank][0].GetReadableAddress(),
741 : role);
742 :
743 0 : std::shared_ptr<TransportRemoteAccess> transportPtr;
744 0 : transportPtr.reset(new (std::nothrow) TransportRemoteAccess(
745 0 : tag_, dispatcher_, notifyPool_, accessPara, addrInfos_, deviceLogicId_));
746 0 : CHK_PRT_RET(!transportPtr, HCCL_ERROR("[Init][DestTransport]InitDestTransport failed"), HCCL_E_PTR);
747 :
748 0 : std::unique_lock<std::mutex> remoteTransportMapLock(remoteTransportMapLock_);
749 0 : remoteTransportMap_.insert(std::make_pair(dstRank, transportPtr));
750 0 : remoteTransportMapLock.unlock();
751 :
752 0 : CHK_RET(transportPtr->Init());
753 0 : *getThreadStatus = 0;
754 0 : return HCCL_SUCCESS;
755 0 : }
756 :
757 : // 根据socket handle，获取本device所使用的网口IP
758 0 : HcclResult CommRemoteAccess::GetNicByHandle(const SocketHandle socketHandle, HcclIpAddress& nicIp)
759 : {
760 0 : for (auto it = raResourceInfo_.nicSocketMap.begin(); it != raResourceInfo_.nicSocketMap.end(); it++) {
761 0 : if (it->second.nicSocketHandle == socketHandle) {
762 0 : nicIp = it->first;
763 0 : return HCCL_SUCCESS;
764 : }
765 : }
766 :
767 0 : HCCL_ERROR("[Get][NicByHandle]current socket handle error");
768 0 : return HCCL_E_NOT_FOUND;
769 : }
770 :
771 0 : HcclResult CommRemoteAccess::SetAccessPara(
772 : u32 role, const HcclIpAddress& nicIp, u32 dstRank, FdHandle socketFdhandle, RemoteAccessPara& accessPara)
773 : {
774 0 : accessPara.role = role;
775 0 : accessPara.localIp = nicIp;
776 0 : accessPara.localRank = rank_;
777 0 : accessPara.remoteRank = dstRank;
778 0 : accessPara.socketFdhandle = socketFdhandle;
779 0 : accessPara.raResourceInfo = raResourceInfo_;
780 :
781 : // 获取 nicSocketHandle
782 0 : auto itSocket = raResourceInfo_.nicSocketMap.find(nicIp);
783 0 : if (itSocket == raResourceInfo_.nicSocketMap.end()) {
784 0 : HCCL_ERROR(
785 : "[Set][AccessPara]In get nic handle, can not find socket handle, handle size[%u], local ip[%s]",
786 : raResourceInfo_.nicSocketMap.size(), nicIp.GetReadableAddress());
787 0 : return HCCL_E_PARA;
788 : }
789 0 : accessPara.nicSocketHandle = itSocket->second.nicSocketHandle;
790 0 : CHK_PTR_NULL(accessPara.nicSocketHandle);
791 :
792 0 : accessPara.nicRdmaHandle = itSocket->second.nicRdmaHandle;
793 0 : CHK_PTR_NULL(accessPara.nicRdmaHandle);
794 0 : return HCCL_SUCCESS;
795 : }
796 : } // namespace hccl
|
