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