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 <thread>
12 : #include <chrono>
13 : #include "securec.h"
14 : #include "adapter_hccp.h"
15 : #include "adapter_tdt.h"
16 : #include "adapter_hal.h"
17 : #include "adapter_rts.h"
18 : #include "adapter_rts_common.h"
19 : #include "dispatcher_task_types.h"
20 : #include "network_manager_pub.h"
21 : #include "externalinput.h"
22 : #include "dlra_function.h"
23 : #include "sal_pub.h"
24 : #include "ping_mesh.h"
25 : #include "hccp_ping.h"
26 : #include "hccp_ctx.h"
27 : #include "local_ub_rma_buffer.h"
28 : #include "orion_adapter_hccp.h"
29 :
30 : namespace hccl {
31 : constexpr int HOP_MAX_TIMES = 64; // 最大跳数
32 : constexpr int LOG_CHK_TIMES = 50; // 需要打印日志的轮询次数
33 : constexpr u32 WR_DEPTH_MULTIPLE = 4; // wr深度扩展倍数
34 : constexpr u32 BYTE_PER_TARGET_DEFAULT = 2048; // 记录每个target的result时默认需要的buffersize大小
35 : constexpr u32 BYTE_IPV4_SHIFT_IN_GID = 12; // ipv4地址相比于gid首地址的偏移值
36 : constexpr u32 RPING_PAYLOAD_REFILL_LEN = 136; // A2/3payload头需要重填的长度
37 : constexpr u32 RPING_PAYLOAD_RSVD_LEN = 44; // A2/3payload头需要清零的长度
38 : constexpr u32 RPING_PAYLOAD_UB_HEAD_LEN = 256; // A5payload头需要填充的内存大小
39 : constexpr u32 RPING_PAYLOAD_UB_TIME_LEN = 64; // A5payload头需要填充的times
40 : constexpr u32 QPINFO_UB_KEY_LEN =28; //jetty存储
41 :
42 :
43 : enum class RpingInitState {
44 : HCCL_INIT_SUCCESS,
45 : HCCL_TSD_NEED_CLOSE,
46 : HCCL_RA_NEED_DEINIT,
47 : HCCL_RAPING_NEED_DEINIT,
48 : HCCL_NET_NEED_CLOSE,
49 : RESERVED
50 : };
51 :
52 21 : PingMesh::PingMesh()
53 21 : {}
54 :
55 20 : PingMesh::~PingMesh()
56 : {
57 20 : if (!isDeinited_) { // 如果没有主动释放过资源，析构时需要释放一下
58 20 : HccnRpingDeinit(deviceLogicId_);
59 : }
60 20 : }
61 :
62 : static bool isInitialized = false; // 标记是否已经初始化
63 : static std::mutex ubTokenMutex;
64 :
65 1 : inline HcclResult GetUbToken(u32 devicePhyId, u32* client_qp_token, u32* client_seg_token,
66 : u32* server_qp_token, u32* server_seg_token)
67 : {
68 1 : std::lock_guard<std::mutex> lock(ubTokenMutex);
69 1 : if (!isInitialized) {
70 1 : u32 devPhyId = devicePhyId;
71 1 : struct RaInfo raInfo = {};
72 1 : raInfo.mode = HrtNetworkMode::HDC;
73 1 : raInfo.phyId = devPhyId;
74 1 : HcclResult ret = hrtRaGetSecRandom(&raInfo, client_qp_token);
75 1 : if (ret != HCCL_SUCCESS) {
76 0 : HCCL_ERROR("get hrtRaGetSecRandom client_qp_token failed, ret:%d", ret);
77 0 : return ret;
78 : }
79 1 : ret = hrtRaGetSecRandom(&raInfo, client_seg_token);
80 1 : if (ret != HCCL_SUCCESS) {
81 0 : HCCL_ERROR("get hrtRaGetSecRandom client_seg_token failed, ret:%d", ret);
82 0 : return ret;
83 : }
84 1 : ret = hrtRaGetSecRandom(&raInfo, server_qp_token);
85 1 : if (ret != HCCL_SUCCESS) {
86 0 : HCCL_ERROR("get hrtRaGetSecRandom server_qp_token failed, ret:%d", ret);
87 0 : return ret;
88 : }
89 1 : ret = hrtRaGetSecRandom(&raInfo, server_seg_token);
90 1 : if (ret != HCCL_SUCCESS) {
91 0 : HCCL_ERROR("get hrtRaGetSecRandom server_seg_token failed, ret:%d", ret);
92 0 : return ret;
93 : }
94 1 : isInitialized = true;
95 : }
96 :
97 1 : return HCCL_SUCCESS;
98 1 : }
99 :
100 20 : bool IsSupportHCCLV2(const char *socNamePtr)
101 : {
102 20 : std::string targetChipVerStr = socNamePtr;
103 20 : HCCL_DEBUG("[%s]SocVersion = %s.", __func__, targetChipVerStr.c_str());
104 20 : if (targetChipVerStr.find("Ascend950") != std::string::npos) {
105 2 : return true;
106 : }
107 :
108 18 : if (targetChipVerStr.find("Ascend910_96") != std::string::npos
109 18 : || targetChipVerStr.find("Ascend960") != std::string::npos
110 36 : || targetChipVerStr.find("ascend960") != std::string::npos) {
111 0 : return true;
112 : }
113 :
114 18 : return false;
115 20 : }
116 :
117 18 : HcclResult GetAddrType(u32 *addrType)
118 : {
119 18 : CHK_PTR_NULL(addrType);
120 18 : const char *socNamePtr = aclrtGetSocName();
121 18 : CHK_PTR_NULL(socNamePtr);
122 18 : if (IsSupportHCCLV2(socNamePtr)) {
123 0 : *addrType = HCCN_RPING_ADDR_TYPE_EID;
124 : } else {
125 18 : *addrType = HCCN_RPING_ADDR_TYPE_IP;
126 : }
127 18 : return HCCL_SUCCESS;
128 : }
129 :
130 19 : inline HcclResult UninitStateCheck(RpingState nextState)
131 : {
132 19 : HcclResult ret = HCCL_SUCCESS;
133 19 : switch(nextState) {
134 0 : case RpingState::UNINIT:
135 0 : HCCL_INFO("[HCCN][UninitStateCheck]Device is uninited, does not need uninit.");
136 0 : break;
137 16 : case RpingState::INITED:
138 16 : break;
139 1 : case RpingState::READY:
140 1 : HCCL_ERROR("[HCCN][UninitStateCheck]Device is not inited yet.");
141 1 : ret = HCCL_E_NOT_SUPPORT;
142 1 : break;
143 1 : case RpingState::RUN:
144 1 : HCCL_ERROR("[HCCN][UninitStateCheck]Device is not inited yet.");
145 1 : ret = HCCL_E_NOT_SUPPORT;
146 1 : break;
147 1 : case RpingState::STOP:
148 1 : HCCL_ERROR("[HCCN][UninitStateCheck]Device is not inited yet.");
149 1 : ret = HCCL_E_NOT_SUPPORT;
150 1 : break;
151 0 : default:
152 0 : HCCL_ERROR("[HCCN][UninitStateCheck]Undefined behavior.");
153 0 : ret = HCCL_E_NOT_SUPPORT;
154 0 : break;
155 : }
156 :
157 19 : return ret;
158 : }
159 :
160 16 : inline HcclResult InitedStateCheck(RpingState nextState)
161 : {
162 16 : HcclResult ret = HCCL_SUCCESS;
163 16 : switch(nextState) {
164 9 : case RpingState::UNINIT:
165 9 : break;
166 0 : case RpingState::INITED:
167 0 : HCCL_INFO("[HCCN][InitedStateCheck]Device is inited already.");
168 0 : break;
169 5 : case RpingState::READY:
170 5 : break;
171 1 : case RpingState::RUN:
172 1 : HCCL_ERROR("[HCCN][InitedStateCheck]Device is not ready.");
173 1 : ret = HCCL_E_NOT_SUPPORT;
174 1 : break;
175 1 : case RpingState::STOP:
176 1 : HCCL_ERROR("[HCCN][InitedStateCheck]Device is not ready.");
177 1 : ret = HCCL_E_NOT_SUPPORT;
178 1 : break;
179 0 : default:
180 0 : HCCL_ERROR("[HCCN][InitedStateCheck]Undefined behavior.");
181 0 : ret = HCCL_E_NOT_SUPPORT;
182 0 : break;
183 : }
184 :
185 16 : return ret;
186 : }
187 :
188 9 : inline HcclResult ReadyStateCheck(RpingState nextState)
189 : {
190 9 : HcclResult ret = HCCL_SUCCESS;
191 9 : switch(nextState) {
192 2 : case RpingState::UNINIT:
193 2 : break;
194 2 : case RpingState::INITED:
195 2 : break;
196 3 : case RpingState::READY:
197 3 : break;
198 1 : case RpingState::RUN:
199 1 : break;
200 1 : case RpingState::STOP:
201 1 : HCCL_ERROR("[HCCN][ReadyStateCheck]Device has not run tasks yet.");
202 1 : ret = HCCL_E_NOT_SUPPORT;
203 1 : break;
204 0 : default:
205 0 : HCCL_ERROR("[HCCN][ReadyStateCheck]Undefined behavior.");
206 0 : ret = HCCL_E_NOT_SUPPORT;
207 0 : break;
208 : }
209 :
210 9 : return ret;
211 : }
212 :
213 4 : inline HcclResult RunStateCheck(RpingState nextState)
214 : {
215 4 : HcclResult ret = HCCL_SUCCESS;
216 4 : switch(nextState) {
217 3 : case RpingState::UNINIT:
218 3 : HCCL_WARNING("[HCCN][RunStateCheck]Make sure the task is finished and result has already gotten.");
219 3 : break;
220 0 : case RpingState::INITED:
221 0 : break;
222 0 : case RpingState::READY:
223 0 : break;
224 0 : case RpingState::RUN:
225 0 : break;
226 1 : case RpingState::STOP:
227 1 : break;
228 0 : default:
229 0 : HCCL_ERROR("[HCCN][RunStateCheck]Undefined behavior.");
230 0 : ret = HCCL_E_NOT_SUPPORT;
231 0 : break;
232 : }
233 :
234 4 : return ret;
235 : }
236 :
237 1 : inline HcclResult StopStateCheck(RpingState nextState)
238 : {
239 1 : HcclResult ret = HCCL_SUCCESS;
240 1 : switch(nextState) {
241 1 : case RpingState::UNINIT:
242 1 : break;
243 0 : case RpingState::INITED:
244 0 : break;
245 0 : case RpingState::READY:
246 0 : break;
247 0 : case RpingState::RUN:
248 0 : break;
249 0 : case RpingState::STOP:
250 0 : HCCL_WARNING("[HCCN][StopStateCheck]Task is stopped.");
251 0 : break;
252 0 : default:
253 0 : HCCL_ERROR("[HCCN][StopStateCheck]Undefined behavior.");
254 0 : ret = HCCL_E_NOT_SUPPORT;
255 0 : break;
256 : }
257 :
258 1 : return ret;
259 : }
260 :
261 49 : inline HcclResult RpingstateCheck(RpingState currState, RpingState nextState)
262 : {
263 49 : HcclResult ret = HCCL_SUCCESS;
264 49 : switch(currState) {
265 19 : case RpingState::UNINIT:
266 19 : ret = UninitStateCheck(nextState);
267 19 : break;
268 16 : case RpingState::INITED:
269 16 : ret = InitedStateCheck(nextState);
270 16 : break;
271 9 : case RpingState::READY:
272 9 : ret = ReadyStateCheck(nextState);
273 9 : break;
274 4 : case RpingState::RUN:
275 4 : ret = RunStateCheck(nextState);
276 4 : break;
277 1 : case RpingState::STOP:
278 1 : ret = StopStateCheck(nextState);
279 1 : break;
280 0 : default:
281 0 : HCCL_ERROR("[HCCN][RpingstateCheck]Current state doesn't exist.");
282 0 : ret = HCCL_E_NOT_SUPPORT;
283 0 : break;
284 : }
285 :
286 49 : return ret;
287 : }
288 :
289 : const std::string extPam[TSD_EXT_PARA_NUM] = {std::string("--hdcType=" + std::to_string(HDC_SERVICE_TYPE_RDMA_V2)),
290 : std::string("--whiteListStatus=" + std::to_string(WHITE_LIST_CLOSE))};
291 16 : inline void TsdProcessOpenInit(rtNetServiceOpenArgs &openArgs, rtProcExtParam *extParam)
292 : {
293 48 : for (u32 i = 0; i < TSD_EXT_PARA_NUM; i++) {
294 32 : extParam[i].paramInfo = extPam[i].c_str();
295 32 : extParam[i].paramLen = extPam[i].size();
296 : }
297 16 : openArgs.extParamList = extParam;
298 16 : openArgs.extParamCnt = TSD_EXT_PARA_NUM;
299 16 : HCCL_INFO("[HCCN]TsdProcessOpenInit extPar0[%s] size[%llu], extPar1[%s] size[%llu]",
300 : extParam[0].paramInfo, extParam[0].paramLen, extParam[1].paramInfo, extParam[1].paramLen);
301 16 : }
302 :
303 14 : inline void RpingRoceAttrInit(u32 deviceId, HcclIpAddress ipAddr, u32 port, u32 nodeNum, u32 bufferSize, u32 sl, u32 tc,
304 : PingInitAttr &initAttr)
305 : {
306 14 : u32 maxWrDepth = nodeNum * WR_DEPTH_MULTIPLE;
307 14 : maxWrDepth = (maxWrDepth > DEFAULT_OPBASE_MAX_SEND_WR) ? DEFAULT_OPBASE_MAX_SEND_WR : maxWrDepth;
308 :
309 14 : initAttr.version = 0; // 暂时无用，默认给0
310 14 : initAttr.mode = NETWORK_OFFLINE; // net work mode 枚举值
311 14 : initAttr.dev.rdma.phyId = deviceId;
312 14 : initAttr.dev.rdma.family = ipAddr.GetFamily(); // AF_INET(ipv4) or AF_INET6(ipv6)
313 14 : initAttr.dev.rdma.localIp.addr = ipAddr.GetBinaryAddress().addr;
314 14 : initAttr.dev.rdma.localIp.addr6 = ipAddr.GetBinaryAddress().addr6;
315 14 : initAttr.bufferSize = bufferSize == 0 ? (maxWrDepth * BYTE_PER_TARGET_DEFAULT) : bufferSize; // 发送接收缓存区大小
316 14 : initAttr.protocol = PROTOCOL_RDMA; // pingmesh支持兼容UB驱动，新增protocol字段
317 :
318 : // client的初始化信息
319 14 : initAttr.client.rdma.cqAttr.sendCqDepth = maxWrDepth;
320 14 : initAttr.client.rdma.cqAttr.recvCqDepth = maxWrDepth;
321 14 : initAttr.client.rdma.cqAttr.sendCqCompVector = 0; // 一组cqe组成的集合，这里给0
322 14 : initAttr.client.rdma.cqAttr.recvCqCompVector = 1; // 一组cqe组成的集合，这里给1
323 14 : initAttr.client.rdma.qpAttr.cap.maxSendWr = maxWrDepth;
324 14 : initAttr.client.rdma.qpAttr.cap.maxRecvWr = maxWrDepth;
325 14 : initAttr.client.rdma.qpAttr.cap.maxSendSge = DEFAULT_MAX_SEND_SGE;
326 14 : initAttr.client.rdma.qpAttr.cap.maxRecvSge = DEFAULT_MAX_RECV_SGE;
327 14 : initAttr.client.rdma.qpAttr.cap.maxInlineData = DEFAULT_MAX_INLINE_DATA;
328 14 : initAttr.client.rdma.qpAttr.udpSport = 0;
329 :
330 : // server的初始化信息
331 14 : initAttr.server.rdma.cqAttr.sendCqDepth = maxWrDepth;
332 14 : initAttr.server.rdma.cqAttr.recvCqDepth = maxWrDepth;
333 14 : initAttr.server.rdma.cqAttr.sendCqCompVector = 0; // 一组cqe组成的集合，这里给0
334 14 : initAttr.server.rdma.cqAttr.recvCqCompVector = 1; // 一组cqe组成的集合，这里给1
335 14 : initAttr.server.rdma.qpAttr.cap.maxSendWr = maxWrDepth;
336 14 : initAttr.server.rdma.qpAttr.cap.maxRecvWr = maxWrDepth;
337 14 : initAttr.server.rdma.qpAttr.cap.maxSendSge = DEFAULT_MAX_SEND_SGE;
338 14 : initAttr.server.rdma.qpAttr.cap.maxRecvSge = DEFAULT_MAX_RECV_SGE;
339 14 : initAttr.server.rdma.qpAttr.cap.maxInlineData = DEFAULT_MAX_INLINE_DATA;
340 14 : initAttr.server.rdma.qpAttr.udpSport = 0;
341 :
342 : // ip协议信息
343 14 : initAttr.commInfo.version = 0;
344 14 : initAttr.commInfo.rdma.flowLabel = 0;
345 14 : initAttr.commInfo.rdma.hopLimit = HOP_MAX_TIMES;
346 14 : initAttr.commInfo.rdma.qosAttr.sl = sl;
347 14 : initAttr.commInfo.rdma.qosAttr.tc = tc;
348 14 : }
349 :
350 1 : inline HcclResult RpingUbAttrInit(u32 deviceId, HcclIpAddress ipAddr, u32 port, u32 nodeNum, u32 bufferSize, u32 sl, u32 tc,
351 : PingInitAttr &initAttr, std::map<Eid, uint32_t> eidmap)
352 : {
353 1 : u32 maxWrDepth = nodeNum * WR_DEPTH_MULTIPLE;
354 1 : maxWrDepth = (maxWrDepth > DEFAULT_OPBASE_MAX_SEND_WR) ? DEFAULT_OPBASE_MAX_SEND_WR : maxWrDepth;
355 :
356 1 : initAttr.version = 0; // 暂时无用，默认给0
357 1 : initAttr.mode = NETWORK_OFFLINE; // net work mode 枚举值
358 1 : initAttr.ub.phyId = deviceId;
359 1 : if (eidmap.find(ipAddr.GetEid()) == eidmap.end()) {
360 0 : HCCL_ERROR("eidmap don't have input Eid,Input Eid %s", ipAddr.GetEid().Describe().c_str());
361 0 : return HCCL_E_NOT_FOUND;
362 : }
363 1 : initAttr.dev.ub.eidIndex = eidmap.at(ipAddr.GetEid());//从eid_list获取eidIndex
364 1 : u32 ret = memcpy_s(initAttr.dev.ub.eid.raw, sizeof(initAttr.dev.ub.eid.raw),
365 1 : ipAddr.GetEid().raw, sizeof(ipAddr.GetEid().raw));
366 1 : if (ret != 0) {
367 0 : HCCL_ERROR("memcpy_s Eid failed");
368 0 : return HCCL_E_MEMORY;
369 : }
370 1 : initAttr.bufferSize = bufferSize == 0 ? (maxWrDepth * BYTE_PER_TARGET_DEFAULT) : bufferSize; // 发送接收缓存区大小
371 1 : initAttr.protocol = PROTOCOL_UDMA; // pingmesh支持兼容UB驱动，新增protocol字段
372 :
373 : //获取安全随机数
374 : u32 client_qp_token, client_seg_token;
375 : u32 server_qp_token, server_seg_token;
376 1 : HcclResult token_ret = GetUbToken(deviceId, &client_qp_token, &client_seg_token, &server_qp_token, &server_seg_token);
377 1 : CHK_RET(token_ret);
378 : // client的初始化信息
379 1 : initAttr.client.ub.cqAttr.sendCqDepth = maxWrDepth;
380 1 : initAttr.client.ub.cqAttr.recvCqDepth = maxWrDepth;
381 1 : initAttr.client.ub.cqAttr.sendCqCompVector = 0; // 一组cqe组成的集合，这里给0
382 1 : initAttr.client.ub.cqAttr.recvCqCompVector = 1; // 一组cqe组成的集合，这里给1
383 1 : initAttr.client.ub.qpAttr.cap.maxSendWr = maxWrDepth;
384 1 : initAttr.client.ub.qpAttr.cap.maxRecvWr = maxWrDepth;
385 1 : initAttr.client.ub.qpAttr.cap.maxSendSge = DEFAULT_MAX_SEND_SGE;
386 1 : initAttr.client.ub.qpAttr.cap.maxRecvSge = DEFAULT_MAX_RECV_SGE;
387 1 : initAttr.client.ub.qpAttr.cap.maxInlineData = DEFAULT_MAX_INLINE_DATA;
388 1 : initAttr.client.ub.qpAttr.tokenValue = client_qp_token;
389 1 : initAttr.client.ub.segAttr.tokenValue = client_seg_token;
390 :
391 : // server的初始化信息
392 1 : initAttr.server.ub.cqAttr.sendCqDepth = maxWrDepth;
393 1 : initAttr.server.ub.cqAttr.recvCqDepth = maxWrDepth;
394 1 : initAttr.server.ub.cqAttr.sendCqCompVector = 0; // 一组cqe组成的集合，这里给0
395 1 : initAttr.server.ub.cqAttr.recvCqCompVector = 1; // 一组cqe组成的集合，这里给1
396 1 : initAttr.server.ub.qpAttr.cap.maxSendWr = maxWrDepth;
397 1 : initAttr.server.ub.qpAttr.cap.maxRecvWr = maxWrDepth;
398 1 : initAttr.server.ub.qpAttr.cap.maxSendSge = DEFAULT_MAX_SEND_SGE;
399 1 : initAttr.server.ub.qpAttr.cap.maxRecvSge = DEFAULT_MAX_RECV_SGE;
400 1 : initAttr.server.ub.qpAttr.cap.maxInlineData = DEFAULT_MAX_INLINE_DATA;
401 1 : initAttr.server.ub.qpAttr.tokenValue = server_qp_token;
402 1 : initAttr.server.ub.segAttr.tokenValue = server_seg_token;
403 :
404 : // ip协议信息
405 1 : initAttr.commInfo.version = 0;
406 1 : initAttr.commInfo.ub.qosAttr.sl = sl;
407 1 : initAttr.commInfo.ub.qosAttr.tc = tc;
408 1 : return HCCL_SUCCESS;
409 : }
410 : const std::unordered_map<HrtNetworkMode, NetworkMode, std::EnumClassHash> HRT_NETWORK_MODE_MAP
411 : = {{HrtNetworkMode::PEER, NetworkMode::NETWORK_PEER_ONLINE}, {HrtNetworkMode::HDC, NetworkMode::NETWORK_OFFLINE}};
412 :
413 : //add查询eidIndex
414 2 : inline HcclResult RaGetEidMap(std::map<Eid, uint32_t>& eidmap, const HRaInfo &raInfo)
415 : {
416 2 : struct RaInfo info {};
417 2 : u32 num = 0;
418 2 : s32 ret = 0;
419 :
420 2 : auto iter = HRT_NETWORK_MODE_MAP.find(raInfo.mode);
421 2 : if (iter == HRT_NETWORK_MODE_MAP.end()) {
422 0 : HCCL_ERROR("[RaGetEidMap]HRT_NETWORK_MODE_MAP not found mode[%d].", raInfo.mode);
423 0 : return HCCL_E_NOT_FOUND;
424 : }
425 2 : info.mode = iter->second;
426 2 : info.phyId = raInfo.phyId;
427 :
428 2 : ret = hrtRaGetDevEidInfoNum(info, &num);
429 2 : if (ret != 0) {
430 1 : HCCL_ERROR("call RaGetDevEidInfoNum failed, error code = %d.", ret);
431 1 : return HCCL_E_NETWORK; //ra接口是网络相关调用
432 : }
433 1 : if (num == 0) {
434 0 : HCCL_WARNING("call hrtRaGetDevEidInfoNum return num = 0.");
435 0 : return HCCL_SUCCESS;
436 : }
437 :
438 1 : HccpDevEidInfo *infoList = new (std::nothrow) HccpDevEidInfo[num];
439 1 : CHK_PTR_NULL(infoList);
440 1 : ret = hrtRaGetDevEidInfoList(info, infoList, &num);
441 1 : if (ret != 0 || num == 0) {
442 0 : HCCL_ERROR("call RaGetDevEidInfoList failed num = %u, error code = %d.", num, ret);
443 0 : delete[] infoList;
444 0 : return HCCL_E_NETWORK;
445 : }
446 :
447 : //填充map
448 2 : for (u32 i = 0; i < num; i++) {
449 1 : Eid eid;
450 2 : ret = memcpy_s(eid.raw, sizeof(eid.raw),
451 1 : infoList[i].eid.raw, sizeof(infoList[i].eid.raw));
452 1 : if (ret != 0) {
453 0 : HCCL_ERROR("[RaGetEidMap]memcpy_s failed, error code = %d.", ret);
454 0 : delete[] infoList;
455 0 : return HCCL_E_INTERNAL;
456 : }
457 1 : HCCL_RUN_INFO("[RaGetEidMap] eid[%s], eidIndex[%u] get.", eid.Describe().c_str(), infoList[i].eidIndex);
458 1 : eidmap.insert(std::make_pair(eid, infoList[i].eidIndex));
459 : }
460 1 : delete[] infoList;
461 :
462 1 : return HCCL_SUCCESS;
463 : }
464 :
465 0 : inline HcclResult RpingTargetAttrInitWithUb(PingTargetInfo &ubtarget, RpingInput ubinput, PingQpInfo *ubinfo, bool isAddTargetUb)
466 : {
467 0 : ubtarget.remoteInfo.qpInfo.version = ubinfo->version;
468 0 : ubtarget.remoteInfo.qpInfo.ub.size = ubinfo->ub.size;
469 0 : u32 ret = 0;
470 0 : ret = memcpy_s(ubtarget.remoteInfo.qpInfo.ub.key, sizeof(ubtarget.remoteInfo.qpInfo.ub.key),
471 0 : ubinfo->ub.key, QPINFO_UB_KEY_LEN);
472 0 : if (ret != 0) {
473 0 : HCCL_ERROR("[RpingTargetAttrInitWithUb]memcpy_s key failed, error code = %d.", ret);
474 0 : return HCCL_E_INTERNAL;
475 : }
476 0 : ubtarget.remoteInfo.qpInfo.ub.tokenValue = ubinfo->ub.tokenValue;
477 0 : ret = memcpy_s(ubtarget.remoteInfo.eid.raw, sizeof(ubtarget.remoteInfo.eid.raw),
478 0 : ubinput.dip.GetEid().raw, URMA_EID_LEN);
479 0 : if (ret != 0) {
480 0 : HCCL_ERROR("[RpingTargetAttrInitWithUb]memcpy_s eid failed, error code = %d.", ret);
481 0 : return HCCL_E_INTERNAL;
482 : }
483 0 : ubtarget.localInfo.ub.qosAttr.tc = ubinput.tc;
484 0 : ubtarget.localInfo.ub.qosAttr.sl = ubinput.sl;
485 0 : if (!isAddTargetUb) { // 并非添加target的时候调用，不需要拷贝payload信息
486 0 : return HCCL_SUCCESS;
487 : }
488 0 : if (ubinput.len > PING_USER_PAYLOAD_MAX_SIZE) {
489 0 : HCCL_WARNING(
490 : "[HCCN][RpingTargetAttrInit]Payload length is %u, should be less than %u byte.", ubinput.len, PING_USER_PAYLOAD_MAX_SIZE);
491 0 : ubtarget.payload.size = 0;
492 0 : return HCCL_SUCCESS;
493 : }
494 0 : ubtarget.payload.size = ubinput.len;
495 0 : errno_t memRet = memcpy_s(ubtarget.payload.buffer, ubtarget.payload.size, ubinput.payload, ubinput.len);
496 0 : if (memRet != EOK) {
497 0 : HCCL_ERROR("[HCCN][RpingTargetAttrInit]Memcpy ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
498 : memRet, ubtarget.payload.buffer, ubtarget.payload.size, ubinput.payload, ubinput.len);
499 0 : return HCCL_E_MEMORY;
500 : }
501 :
502 0 : return HCCL_SUCCESS;
503 : }
504 :
505 15 : inline HcclResult RpingTargetAttrInit(PingTargetInfo &target, RpingInput input, PingQpInfo *rdmainfo, bool isAddTargetUb)
506 : {
507 15 : target.remoteInfo.qpInfo.version = rdmainfo->version;
508 15 : target.remoteInfo.qpInfo.rdma.gid = rdmainfo->rdma.gid;
509 15 : target.remoteInfo.qpInfo.rdma.qpn = rdmainfo->rdma.qpn;
510 15 : target.remoteInfo.qpInfo.rdma.qkey = rdmainfo->rdma.qkey;
511 15 : target.remoteInfo.ip.addr = input.dip.GetBinaryAddress().addr;
512 15 : target.remoteInfo.ip.addr6 = input.dip.GetBinaryAddress().addr6;
513 15 : target.localInfo.rdma.qosAttr.tc = input.tc;
514 15 : target.localInfo.rdma.qosAttr.sl = input.sl;
515 15 : target.localInfo.rdma.flowLabel = 0;
516 15 : target.localInfo.rdma.hopLimit = HOP_MAX_TIMES;
517 15 : target.localInfo.rdma.udpSport = input.srcPort;
518 15 : if (!isAddTargetUb) { // 并非添加target的时候调用，不需要拷贝payload信息
519 2 : return HCCL_SUCCESS;
520 : }
521 13 : if (input.len > PING_USER_PAYLOAD_MAX_SIZE) {
522 0 : HCCL_WARNING(
523 : "[HCCN][RpingTargetAttrInit]Payload length is %u, should be less than %u byte.", input.len, PING_USER_PAYLOAD_MAX_SIZE);
524 0 : target.payload.size = 0;
525 0 : return HCCL_SUCCESS;
526 : }
527 13 : target.payload.size = input.len;
528 13 : errno_t memRet = memcpy_s(target.payload.buffer, target.payload.size, input.payload, input.len);
529 13 : if (memRet != EOK) {
530 1 : HCCL_ERROR("[HCCN][RpingTargetAttrInit]Memcpy ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
531 : memRet, target.payload.buffer, target.payload.size, input.payload, input.len);
532 1 : return HCCL_E_MEMORY;
533 : }
534 :
535 12 : return HCCL_SUCCESS;
536 : }
537 :
538 2 : HcclResult PingMesh::RpingResultInfoInit(PingTargetResult *resultInfo,
539 : UniversalConcurrentMap<std::string, PingQpInfo> &rdmaInfoMaps, RpingInput *input, u32 targetNum)
540 : {
541 2 : u32 addressType = 0;
542 2 : HcclResult addrTypeRet = GetAddrType(&addressType);
543 2 : if (addrTypeRet != HCCL_SUCCESS) {
544 0 : HCCL_ERROR("[RpingResultInfoInit]GetAddrType Fail ret %d", addrTypeRet);
545 0 : return HCCL_E_PARA;
546 : }
547 22 : for (u32 i = 0; i < targetNum; i++) {
548 40 : if (!rdmaInfoMaps.Find(std::string(input[i].dip.GetReadableIP())).second) {
549 0 : HCCL_WARNING("[HCCN][RpingResultInfoInit]Target[%s] info doesn't exist.", input[i].dip.GetReadableIP());
550 0 : continue;
551 : }
552 20 : PingQpInfo *rdmainfo = &rdmaInfoMaps[std::string(input[i].dip.GetReadableIP())];
553 20 : if (addressType == HCCN_RPING_ADDR_TYPE_IP) {
554 20 : resultInfo[i].remoteInfo.ip.addr = input[i].dip.GetBinaryAddress().addr;
555 20 : resultInfo[i].remoteInfo.ip.addr6 = input[i].dip.GetBinaryAddress().addr6;
556 20 : resultInfo[i].remoteInfo.qpInfo.version = 0;
557 20 : resultInfo[i].remoteInfo.qpInfo.rdma.gid = rdmainfo->rdma.gid;
558 20 : resultInfo[i].remoteInfo.qpInfo.rdma.qpn = rdmainfo->rdma.qpn;
559 20 : resultInfo[i].remoteInfo.qpInfo.rdma.qkey = rdmainfo->rdma.qkey;
560 : }
561 20 : const char *socNamePtr = aclrtGetSocName();
562 20 : CHK_PTR_NULL(socNamePtr);
563 20 : if (addressType == HCCN_RPING_ADDR_TYPE_EID && IsSupportHCCLV2(socNamePtr)) {
564 0 : u32 ret = 0;
565 0 : ret = memcpy_s(resultInfo[i].remoteInfo.eid.raw, sizeof(resultInfo[i].remoteInfo.eid.raw),
566 0 : input[i].dip.GetEid().raw, URMA_EID_LEN);
567 0 : if (ret != 0) {
568 0 : HCCL_ERROR("[RpingResultInfoInit]memcpy_s eid failed, error code = %d.", ret);
569 0 : return HCCL_E_INTERNAL;
570 : }
571 0 : resultInfo[i].remoteInfo.qpInfo.version = 0;
572 0 : resultInfo[i].remoteInfo.qpInfo.ub.size = rdmainfo->ub.size;
573 0 : ret = memcpy_s(resultInfo[i].remoteInfo.qpInfo.ub.key, sizeof(resultInfo[i].remoteInfo.qpInfo.ub.key),
574 0 : rdmainfo->ub.key, QPINFO_UB_KEY_LEN);
575 0 : if (ret != 0) {
576 0 : HCCL_ERROR("[RpingResultInfoInit]memcpy_s key failed, error code = %d.", ret);
577 0 : return HCCL_E_INTERNAL;
578 : }
579 0 : resultInfo[i].remoteInfo.qpInfo.ub.tokenValue = rdmainfo->ub.tokenValue;
580 : }
581 20 : HCCL_INFO("[HCCN][RpingResultInfoInit]Target[%s] info init success.", input[i].dip.GetReadableIP());
582 :
583 : }
584 2 : return HCCL_SUCCESS;
585 : }
586 :
587 2 : inline void GetResultFromReturnValue(PingTargetResult *resultInfo, RpingOutput *output, u32 targetNum)
588 : {
589 22 : for (u32 i = 0; i < targetNum; i++) {
590 20 : output[i].state = resultInfo[i].result.state;
591 20 : output[i].txPkt = resultInfo[i].result.summary.sendCnt;
592 20 : output[i].rxPkt = resultInfo[i].result.summary.recvCnt;
593 20 : output[i].minRTT = resultInfo[i].result.summary.rttMin;
594 20 : output[i].maxRTT = resultInfo[i].result.summary.rttMax;
595 20 : output[i].avgRTT = resultInfo[i].result.summary.rttAvg;
596 : }
597 2 : }
598 :
599 3 : inline void LogRecordbyTimes(int &count)
600 : {
601 : // 日志过滤, 50次才打印一次
602 3 : if (count % LOG_CHK_TIMES == 0) {
603 2 : HCCL_DEBUG("[HCCN][LogRecordbyTimes]socket is connecting...");
604 : }
605 3 : count++;
606 3 : }
607 :
608 1 : inline void RemoveMapInfo(RpingInput *input, u32 targetNum,
609 : std::map<std::string, std::shared_ptr<HcclSocket>> &socketMaps,
610 : UniversalConcurrentMap<std::string, PingQpInfo> &rdmaInfoMaps,
611 : UniversalConcurrentMap<std::string, u32> &payloadLenMap)
612 : {
613 2 : for (u32 i = 0; i < targetNum; i++) {
614 2 : socketMaps.erase(std::string(input[i].dip.GetReadableIP()));
615 2 : rdmaInfoMaps.Erase(std::string(input[i].dip.GetReadableIP()));
616 2 : payloadLenMap.Erase(std::string(input[i].dip.GetReadableIP()));
617 : }
618 1 : }
619 :
620 17 : HcclResult PingMesh::RpingSendInitInfo(u32 deviceId, u32 port, HcclIpAddress ipAddr, PingInitInfo initInfo,
621 : std::shared_ptr<HcclSocket> socket)
622 : {
623 : // 给当前线程添加名字
624 17 : SetThreadName("Hccl_PingMesh");
625 : // 等待client端发送的建链请求
626 17 : HcclIpAddress remoteIp = HcclIpAddress();
627 17 : std::string tag = "PingMesh" + std::string(ipAddr.GetReadableIP());
628 17 : HCCL_INFO("[HCCN][RpingSendInitInfo]socket tag[%s].", tag.c_str());
629 : // 持续在后台等待建链，保证可以处理多个client端的建链请求
630 : while (true) {
631 59 : HcclSocket realSocket(tag, netCtx_, remoteIp, 0, HcclSocketRole::SOCKET_ROLE_SERVER);
632 59 : CHK_RET(realSocket.Init());
633 59 : int count = 0; // 轮询计数
634 : while (true) {
635 62 : bool isStop = connThreadStop_.load();
636 62 : if (isStop == true) {
637 10 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
638 10 : HCCL_INFO("[HCCN][RpingSendInitInfo]Device[%u] stop waiting connect.", deviceId);
639 10 : break;
640 : }
641 52 : HcclSocketStatus status = realSocket.GetStatus();
642 52 : if (status == HcclSocketStatus::SOCKET_OK) {
643 0 : HCCL_DEBUG("[HCCN][RpingSendInitInfo]socket is established. localIp[%s], remoteIp[%s]",
644 : realSocket.GetLocalIp().GetReadableIP(), realSocket.GetRemoteIp().GetReadableIP());
645 0 : break;
646 52 : } else if (status == HcclSocketStatus::SOCKET_CONNECTING) {
647 3 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
648 3 : LogRecordbyTimes(count);
649 3 : continue;
650 49 : } else if (status == HcclSocketStatus::SOCKET_TIMEOUT) {
651 0 : HCCL_WARNING("[HCCN][RpingSendInitInfo]socket connect timeout.");
652 0 : break;
653 : } else {
654 49 : HCCL_WARNING("[HCCN][RpingSendInitInfo]socket connect failed.");
655 49 : break;
656 : }
657 3 : }
658 :
659 : // 判断否需要中止线程
660 59 : bool isStop = connThreadStop_.load();
661 59 : if (isStop == true) {
662 17 : HCCL_INFO("[HCCN][RpingSendInitInfo]Device[%u] background thread stopped.", deviceId);
663 17 : break;
664 : }
665 :
666 : // 建链成功，发送rping初始化信息
667 42 : u64 sendSize = sizeof(initInfo);
668 42 : CHK_RET(realSocket.Send(&initInfo, sendSize));
669 42 : HCCL_INFO("[HCCN][RpingSendInitInfo]Device[%u] rdma info send success.", deviceId);
670 101 : }
671 :
672 17 : return HCCL_SUCCESS;
673 17 : }
674 :
675 2 : HcclResult PingMesh::RpingRecvTargetInfo(void *clientNetCtx, u32 port, HcclIpAddress ipAddr, PingInitInfo &recvInfo, u32 timeout)
676 : {
677 : // 确认是否添加过该IP
678 2 : std::unique_lock<std::mutex> lock(socketMapsMtx_);
679 6 : if (socketMaps_.find(std::string(ipAddr.GetReadableIP())) != socketMaps_.end()) {
680 1 : HCCL_WARNING("[HCCN][RpingRecvTargetInfo]IP address[%s] has already exist.", ipAddr.GetReadableIP());
681 1 : return HCCL_SUCCESS;
682 : }
683 1 : lock.unlock();
684 : // socket建链 这里的建链流程与init里的侦听动作应当使用同一套接口
685 1 : std::string tag = "PingMesh" + std::string(ipAddr.GetReadableIP());
686 1 : HCCL_INFO("[HCCN][RpingRecvTargetInfo]socket tag[%s].", tag.c_str());
687 1 : std::shared_ptr<HcclSocket> socket = nullptr;
688 1 : EXCEPTION_CATCH(
689 : (socket = std::make_shared<HcclSocket>(tag, clientNetCtx, ipAddr, port, HcclSocketRole::SOCKET_ROLE_CLIENT)),
690 : return HCCL_E_PTR);
691 1 : CHK_SMART_PTR_NULL(socket);
692 1 : CHK_RET(socket->Init());
693 1 : CHK_RET(socket->Connect());
694 1 : auto startTime = std::chrono::steady_clock::now();
695 : while (true) {
696 1 : auto endTime = std::chrono::steady_clock::now();
697 : // 计算毫秒差值并转为u32
698 1 : auto duration_ms = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime);
699 1 : u32 ms = static_cast<u32>(duration_ms.count());
700 1 : if (ms >= timeout) {
701 0 : HCCL_ERROR("[HCCN][RpingRecvTargetInfo]Get socket timeout! cost time [%u ms], timeout [%u ms]", ms, timeout);
702 0 : socket->SetStatus(HcclSocketStatus::SOCKET_TIMEOUT);
703 0 : return HCCL_E_TIMEOUT;
704 : }
705 :
706 1 : auto status = socket->GetStatus();
707 1 : if (status == HcclSocketStatus::SOCKET_CONNECTING) {
708 0 : SaluSleep(ONE_MILLISEC);
709 0 : HCCL_INFO("[HCCN][RpingRecvTargetInfo]connecting to server [%s] port [%u]", ipAddr.GetReadableIP(), port);
710 0 : continue;
711 1 : } else if (status != HcclSocketStatus::SOCKET_OK) {
712 0 : HCCL_ERROR("[HCCN][RpingRecvTargetInfo]Get socket failed, ret [%d]", status);
713 0 : return HCCL_E_TCP_CONNECT;
714 : } else {
715 1 : HCCL_INFO("[HCCN][RpingRecvTargetInfo]Get socket success with server [%s] port [%u]",
716 : ipAddr.GetReadableIP(), port);
717 1 : break;
718 : }
719 0 : }
720 :
721 : // 接收发送的信息
722 1 : u32 recvBufLen = sizeof(recvInfo);
723 1 : CHK_RET(socket->Recv(&recvInfo, recvBufLen));
724 1 : HCCL_INFO("[HCCN][RpingRecvTargetInfo]Server[%s] info received success.", ipAddr.GetReadableIP());
725 :
726 : // 记录socket
727 1 : lock.lock();
728 2 : socketMaps_.insert({std::string(ipAddr.GetReadableIP()), socket});
729 :
730 1 : return HCCL_SUCCESS;
731 2 : }
732 :
733 0 : inline RpingLinkState ConvertHcclSocketStatus(HcclSocketStatus socketStatus)
734 : {
735 0 : RpingLinkState status = RpingLinkState::DISCONNECTED;
736 0 : switch (socketStatus) {
737 0 : case HcclSocketStatus::SOCKET_INIT:
738 0 : status = RpingLinkState::DISCONNECTED;
739 0 : break;
740 0 : case HcclSocketStatus::SOCKET_OK:
741 0 : status = RpingLinkState::CONNECTED;
742 0 : break;
743 0 : case HcclSocketStatus::SOCKET_TIMEOUT:
744 0 : status = RpingLinkState::TIMEOUT;
745 0 : break;
746 0 : case HcclSocketStatus::SOCKET_CONNECTING:
747 0 : status = RpingLinkState::CONNECTING;
748 0 : break;
749 0 : default:
750 0 : status = RpingLinkState::ERROR;
751 0 : break;
752 : }
753 0 : return status;
754 : }
755 :
756 16 : HcclResult PingMesh::HccnRaInit(u32 deviceId)
757 : {
758 16 : RaInitConfig config = { devicePhyId_, static_cast<u32>(NICDeployment::NIC_DEPLOYMENT_DEVICE),
759 16 : HDC_SERVICE_TYPE_RDMA_V2 };
760 16 : u32 rpingInterfaceVersion = 0;
761 16 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).PingMeshRaPingInit(deviceLogicId_, devicePhyId_, &config));
762 16 : CHK_RET(hrtRaGetInterfaceVersion(devicePhyId_, RPING_INTERFACE_OPCODE, &rpingInterfaceVersion));
763 16 : if (rpingInterfaceVersion < RPING_INTERFACE_VERSION) {
764 0 : HCCL_ERROR("[HCCN][HccnRpingInit]this package[%u] does not support rpingInterface for device.",
765 : rpingInterfaceVersion);
766 0 : return HCCL_E_NOT_SUPPORT;
767 : }
768 16 : HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] init hccp success.", deviceId);
769 16 : return HCCL_SUCCESS;
770 : }
771 :
772 16 : HcclResult PingMesh::HccnCloseSubProc(u32 deviceId)
773 : {
774 16 : hrtCloseNetService();
775 16 : HCCL_INFO("[HCCN][HccnCloseSubProc]Device[%u] close hccp process success.", deviceId);
776 16 : return HCCL_SUCCESS;
777 : }
778 :
779 15 : HcclResult PingMesh::StartSocketThread(u32 deviceId, HcclIpAddress ipAddr, u32 port)
780 : {
781 15 : socket_ = std::make_shared<HcclSocket>(netCtx_, port);
782 : // 初始化socket并启动侦听
783 15 : CHK_RET(socket_->Init());
784 15 : CHK_RET(SetTcpMode(true));
785 15 : CHK_RET(socket_->Listen());
786 15 : HCCL_INFO("[HCCN][StartSocketThread]Device[%u] starts listen port[%u].", deviceId, port);
787 : // 等待客户端建链
788 15 : connThread_.reset(new (std::nothrow)
789 15 : std::thread(&PingMesh::RpingSendInitInfo, this, deviceId, port, ipAddr, initInfo_, socket_));
790 15 : CHK_SMART_PTR_NULL(connThread_);
791 15 : return HCCL_SUCCESS;
792 : }
793 :
794 16 : HcclResult PingMesh::HccnSupportedAndGetphyid(u32 deviceId, LinkType netMode)
795 : {
796 16 : if (netMode != LinkType::LINK_ROCE && netMode != LinkType::LINK_UB) {
797 0 : HCCL_ERROR("[HCCN][HccnSupportedAndGetphyid]only support ROCE or UB mode.");
798 0 : return HCCL_E_NOT_SUPPORT;
799 : }
800 : // 获取并验证设备物理id
801 16 : deviceLogicId_ = deviceId;
802 16 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(deviceLogicId_), devicePhyId_));
803 16 : if (deviceId != static_cast<u32>(deviceLogicId_)) {
804 0 : HCCL_ERROR("[HCCN][HccnSupportedAndGetphyid]Input device logicId[%u] don't match real logicId[%d].", deviceId, deviceLogicId_);
805 0 : return HCCL_E_PARA;
806 : }
807 16 : HCCL_INFO("[HCCN][HccnSupportedAndGetphyid]Device logic id is [%d], phy id is [%u].", deviceLogicId_, devicePhyId_);
808 16 : return HCCL_SUCCESS;
809 : }
810 :
811 16 : HcclResult PingMesh::HccnRpingOpenTsd(u32 deviceId, u32 mode, u32 port, u32 nodeNum, u32 bufferSize, u32 sl, u32 tc)
812 : {
813 : // 判断当前状态
814 16 : CHK_RET(RpingstateCheck(rpingState_, RpingState::INITED));
815 16 : HCCL_DEBUG("[HccnRpingOpenTsd]deviceid %u, mode %u, port %u, nodeNum %u, bufferSize %u, sl %u, tc %u", deviceId, mode,
816 : port, nodeNum, bufferSize, sl, tc);
817 : // 当前只支持RoCE和UB
818 16 : LinkType netMode = static_cast<LinkType>(mode);
819 16 : HcclResult ret = HCCL_SUCCESS;
820 16 : ret = HccnSupportedAndGetphyid(deviceId, netMode);
821 16 : if (ret != HCCL_SUCCESS) {
822 0 : HCCL_ERROR("[HCCN][HccnRpingOpenTsd]HccnSupportedAndGetphyid Failed, deviceId[%u] ret[%d].", deviceId, ret);
823 0 : return HCCL_E_NOT_SUPPORT;
824 : }
825 :
826 : // 拉起hccp进程
827 16 : rtProcExtParam extParam[TSD_EXT_PARA_NUM] {};
828 :
829 : rtNetServiceOpenArgs openArgs;
830 16 : TsdProcessOpenInit(openArgs, extParam);
831 16 : CHK_RET(DlTdtFunction::GetInstance().DlTdtFunctionHeterogInit());
832 :
833 16 : CHK_RET(hrtOpenNetService(&openArgs));
834 16 : HCCL_INFO("[HCCN][HccnRpingOpenTsd]Device[%u] open process success", deviceId);
835 16 : return HCCL_SUCCESS;
836 : }
837 :
838 16 : HcclResult PingMesh::HccnRpingInit(u32 deviceId, u32 mode, HcclIpAddress ipAddr, u32 port, u32 nodeNum, u32 bufferSize,
839 : u32 sl, u32 tc)
840 : {
841 16 : CHK_RET(HccnRpingOpenTsd(deviceId, mode, port, nodeNum, bufferSize, sl, tc));
842 :
843 16 : LinkType netMode = static_cast<LinkType>(mode);
844 16 : RpingInitState status = RpingInitState::HCCL_INIT_SUCCESS;
845 16 : HcclResult ret = HCCL_SUCCESS;
846 16 : void *pingHandle = nullptr;
847 16 : const char *socNamePtr = aclrtGetSocName();
848 : do {
849 : // hccp侧初始化ping mesh资源
850 16 : ret = HccnRaInit(deviceId);
851 16 : if (ret != HCCL_SUCCESS) {
852 0 : status = RpingInitState::HCCL_TSD_NEED_CLOSE;
853 0 : HCCL_ERROR("[HCCN][HccnRpingInit]HccnRaInit fail, deviceId[%u] ret[%d].", deviceId, ret);
854 1 : break;
855 : }
856 16 : PingInitAttr initAttr{};
857 16 : if (netMode == LinkType::LINK_ROCE) {
858 14 : RpingRoceAttrInit(devicePhyId_, ipAddr, port, nodeNum, bufferSize, sl, tc, initAttr);
859 : }
860 16 : if (netMode == LinkType::LINK_UB && IsSupportHCCLV2(socNamePtr)) {
861 2 : HRaInfo info(HrtNetworkMode::HDC, devicePhyId_);
862 2 : std::map<Eid, uint32_t> eidmap;
863 2 : ret = RaGetEidMap(eidmap, info);
864 2 : if (ret != HCCL_SUCCESS) {
865 1 : status = RpingInitState::HCCL_TSD_NEED_CLOSE;
866 1 : HCCL_ERROR("[HccnRpingInit]call ra_get_dev_eid_map failed, devideId[%u], error code =%d.", deviceId, ret);
867 1 : break;
868 : }
869 1 : ret = RpingUbAttrInit(devicePhyId_, ipAddr, port, nodeNum, bufferSize, sl, tc, initAttr, eidmap);
870 1 : if (ret != HCCL_SUCCESS) {
871 0 : status = RpingInitState::HCCL_TSD_NEED_CLOSE;
872 0 : HCCL_ERROR("[HccnRpingInit]RpingUbAttrInit failed, devideId[%u], error code =%d.", deviceId, ret);
873 0 : break;
874 : }
875 2 : }
876 15 : ret = hrtRaPingInit(&initAttr, &initInfo_, &pingHandle);
877 15 : if (ret != HCCL_SUCCESS || pingHandle == nullptr) {
878 0 : status = RpingInitState::HCCL_RA_NEED_DEINIT;
879 0 : HCCL_ERROR("[HCCN][HccnRpingInit]hrtRaPingInit fail, deviceId[%u] ret[%d] pingHandle[%p].", deviceId, ret, pingHandle);
880 0 : break;
881 : }
882 15 : HCCL_INFO("[HCCN][HccnRpingInit]Device[%u] init success.", deviceId);
883 :
884 : // 建链并发送初始化信息
885 15 : ret = HcclNetOpenDev(&netCtx_, NicType::DEVICE_NIC_TYPE, devicePhyId_, deviceLogicId_, ipAddr);
886 15 : if (ret != HCCL_SUCCESS || netCtx_ == nullptr) {
887 0 : status = RpingInitState::HCCL_RAPING_NEED_DEINIT;
888 0 : HCCL_ERROR("[HCCN][HccnRpingInit]HcclNetOpenDev fail, deviceId[%u] ret[%d] netCtx_[%p].", deviceId, ret, netCtx_);
889 0 : break;
890 : }
891 :
892 15 : ret = StartSocketThread(deviceId, ipAddr, port);
893 15 : if (ret != HCCL_SUCCESS) {
894 0 : status = RpingInitState::HCCL_NET_NEED_CLOSE;
895 0 : HCCL_ERROR("[HCCN][HccnRpingInit]StartSocketThread fail, deviceId[%u] port[%d].", deviceId, port);
896 0 : break;
897 : }
898 : } while(0);
899 :
900 16 : switch (status) {
901 15 : case RpingInitState::HCCL_INIT_SUCCESS: break;
902 0 : case RpingInitState::HCCL_NET_NEED_CLOSE:
903 0 : if (netCtx_ != nullptr) {
904 0 : HcclNetCloseDev(netCtx_);
905 0 : netCtx_ = nullptr;
906 : }
907 : [[fallthrough]];
908 : case RpingInitState::HCCL_RAPING_NEED_DEINIT:
909 0 : if (pingHandle != nullptr) {
910 0 : (void)hrtRaPingDeinit(pingHandle);
911 : }
912 : [[fallthrough]];
913 : case RpingInitState::HCCL_RA_NEED_DEINIT:
914 0 : (void)NetworkManager::GetInstance(static_cast<s32>(deviceId)).PingMeshRaPingDeinit();
915 : [[fallthrough]];
916 1 : case RpingInitState::HCCL_TSD_NEED_CLOSE:
917 1 : (void)HccnCloseSubProc(deviceId);
918 : [[fallthrough]];
919 1 : default:
920 1 : HCCL_ERROR("[HCCN][HccnRpingInit]HccnRpingInit ret[%d], status[%d].", ret, status);
921 1 : return ret;
922 : }
923 : // 绑定信息
924 15 : pingHandle_ = pingHandle;
925 15 : rpingState_ = RpingState::INITED;
926 15 : ipAddr_ = ipAddr;
927 15 : isUsePayload_ = bufferSize == 0 ? false : true;
928 :
929 15 : return HCCL_SUCCESS;
930 : }
931 :
932 15 : HcclResult PingMesh::HccnRpingDeinit(u32 deviceId)
933 : {
934 : // 判断当前状态
935 15 : CHK_RET(RpingstateCheck(rpingState_, RpingState::UNINIT));
936 15 : CHK_PRT_RET(rpingState_ == RpingState::UNINIT,
937 : HCCL_WARNING("[HCCN][HccnRpingDeinit]Device[%u] has not inited.", deviceId), HCCL_SUCCESS);
938 15 : HCCL_DEBUG("[HccnRpingDeinit]deviceid %u", deviceId);
939 : // 释放payload内存
940 15 : if (payload_ != nullptr) {
941 2 : delete[] payload_;
942 2 : payload_ = nullptr;
943 : }
944 :
945 : // 手动结束背景线程
946 15 : connThreadStop_.store(true);
947 15 : if (connThread_ != nullptr && connThread_->joinable()) {
948 15 : connThread_->join();
949 15 : HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] end background thread success.", deviceId);
950 : }
951 :
952 : // 清空map
953 15 : std::unique_lock<std::mutex> lock(socketMapsMtx_);
954 18 : for (auto &socket: socketMaps_) {
955 3 : if (socket.second->DeInit() != HCCL_SUCCESS) {
956 0 : HCCL_WARNING("[HCCN][HccnRpingDeinit]socket deinit failed");
957 : }
958 : }
959 15 : socketMaps_.clear();
960 15 : HCCL_INFO("[HCCN][HccnRpingDeinit]Socket map clear.");
961 15 : lock.unlock();
962 15 : rdmaInfoMaps_.Clear();
963 15 : HCCL_INFO("[HCCN][HccnRpingDeinit]Rdma info map clear.");
964 15 : payloadLenMap_.Clear();
965 15 : HCCL_INFO("[HCCN][HccnRpingDeinit]payloadLen map clear.");
966 :
967 : // 关闭socket链路
968 15 : if ((socket_ != nullptr) && (!isSocketClosed_)) {
969 15 : CHK_RET(socket_->DeInit());
970 15 : isSocketClosed_ = true;
971 15 : HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] deinit socket success.", deviceId);
972 : }
973 :
974 : // 释放资源
975 15 : if (netCtx_ != nullptr) {
976 15 : HcclNetCloseDev(netCtx_);
977 15 : netCtx_ = nullptr;
978 : }
979 :
980 15 : if (pingHandle_ == nullptr) {
981 0 : HCCL_WARNING("[HCCN][HccnRpingDeinit]Device[%u] don't need to deinit because it is not inited.", deviceId);
982 0 : return HCCL_SUCCESS;
983 : }
984 15 : CHK_RET(hrtRaPingDeinit(pingHandle_));
985 15 : HCCL_INFO("[HCCN][HccnRpingDeinit]Device[%u] deinit hccp success.", deviceId);
986 :
987 : // 关闭hccp进程
988 15 : CHK_RET(NetworkManager::GetInstance(static_cast<s32>(deviceId)).PingMeshRaPingDeinit());
989 15 : CHK_RET(HccnCloseSubProc(deviceId));
990 15 : isDeinited_ = true;
991 15 : rpingState_ = RpingState::UNINIT;
992 15 : return HCCL_SUCCESS;
993 15 : }
994 :
995 6 : HcclResult PingMesh::HccnTargetAttrInter(u32 targetNumInter, RpingInput *inputInter, HccnRpingAddTargetConfig *configInter,PingTargetInfo *targetInter)
996 : {
997 6 : HcclResult ret = HCCL_SUCCESS;
998 6 : u32 addressType = 0;
999 6 : ret = GetAddrType(&addressType);
1000 6 : if (ret != HCCL_SUCCESS) {
1001 0 : HCCL_ERROR("[HccnTargetAttrInter]GetAddrType Fail ret %d", ret);
1002 0 : return HCCL_E_PARA;
1003 : }
1004 18 : for (u32 i = 0; i < targetNumInter; i++) {
1005 : PingInitInfo recvInfo;
1006 15 : ret = RpingRecvTargetInfo(netCtx_, inputInter[i].port, inputInter[i].dip, recvInfo, configInter->connectTimeout);
1007 15 : if (ret != HCCL_SUCCESS) {
1008 1 : HCCL_ERROR("[HCCN][HccnRpingAddTarget]Target[%s] added failed because of error[%d].",
1009 : inputInter[i].dip.GetReadableIP(), ret);
1010 3 : break;
1011 : }
1012 14 : PingQpInfo *rdmaInfo = &(recvInfo.client);
1013 42 : if (rdmaInfoMaps_.Find(std::string(inputInter[i].dip.GetReadableIP())).second) {
1014 1 : HCCL_RUN_INFO("[HCCN][HccnRpingAddTarget]Target[%s] has already added.", inputInter[i].dip.GetReadableIP());
1015 1 : continue;
1016 : }
1017 26 : rdmaInfoMaps_.Emplace(std::pair<std::string, PingQpInfo>(std::string(inputInter[i].dip.GetReadableIP()), recvInfo.client));
1018 39 : if (payloadLenMap_.Find(std::string(inputInter[i].dip.GetReadableIP())).second) {
1019 0 : HCCL_RUN_INFO("[HCCN][HccnRpingAddTarget]Target[%s] has already added.", inputInter[i].dip.GetReadableIP());
1020 0 : continue;
1021 : }
1022 26 : payloadLenMap_.Emplace(std::pair<std::string, u32>(std::string(inputInter[i].dip.GetReadableIP()), inputInter[i].len));
1023 13 : if (addressType == HCCN_RPING_ADDR_TYPE_IP) {
1024 13 : ret = RpingTargetAttrInit(targetInter[0], inputInter[i], rdmaInfo, true);
1025 : }
1026 13 : const char *socNamePtr = aclrtGetSocName();
1027 13 : CHK_PTR_NULL(socNamePtr);
1028 13 : if (addressType == HCCN_RPING_ADDR_TYPE_EID && IsSupportHCCLV2(socNamePtr)) {
1029 0 : ret = RpingTargetAttrInitWithUb(targetInter[0], inputInter[i], rdmaInfo, true);
1030 : }
1031 13 : if (ret != HCCL_SUCCESS) {
1032 1 : HCCL_ERROR("[HCCN][HccnRpingAddTarget]Target[%s] payload added failed.", inputInter[i].dip.GetReadableIP());
1033 1 : break;
1034 : }
1035 12 : ret = hrtRaPingTargetAdd(pingHandle_, targetInter, 1); // hccp侧只能一个一个处理，因此数组大小固定为1
1036 12 : if (ret != HCCL_SUCCESS) {
1037 1 : HCCL_ERROR("[HCCN][HccnRpingAddTarget]Target[%s] added failed because of error[%d]", inputInter[i].dip.GetReadableIP(), ret);
1038 1 : break;
1039 : }
1040 11 : HCCL_INFO("[HCCN][HccnRpingAddTarget]Target[%s] added success.", inputInter[i].dip.GetReadableIP());
1041 11 : rpingTargetNum_++;
1042 : }
1043 6 : return ret;
1044 : }
1045 :
1046 7 : HcclResult PingMesh::HccnRpingAddTarget(u32 deviceId, u32 targetNum, RpingInput *input, HccnRpingAddTargetConfig *config)
1047 : {
1048 : // 校验入参
1049 7 : CHK_PRT_RET(config == nullptr, HCCL_ERROR("[PingMesh::HccnRpingAddTarget]config is null."), HCCL_E_PARA);
1050 : // 判断当前状态
1051 7 : CHK_RET(RpingstateCheck(rpingState_, RpingState::READY));
1052 : // 调用hccp接口添加目标
1053 6 : if (pingHandle_ == nullptr) {
1054 0 : HCCL_ERROR("[HCCN][HccnRpingAddTarget]Device[%u] cannot add targets because it is not inited.", deviceId);
1055 0 : return HCCL_E_NOT_FOUND;
1056 : }
1057 6 : HCCL_INFO("[HccnRpingAddTarget]deviceId %u, targetNum %u", deviceId, targetNum);
1058 6 : HcclResult ret = HCCL_SUCCESS;
1059 6 : PingTargetInfo target[1] = { {0} }; // hccp侧只能一个一个处理，因此数组大小固定为1
1060 :
1061 6 : ret = HccnTargetAttrInter(targetNum, input, config, target);
1062 6 : if ((ret == HCCL_SUCCESS) && (rpingState_ == RpingState::INITED)) { // 从初始化完成的状态切换到ready to start的状态
1063 2 : rpingState_ = RpingState::READY;
1064 : }
1065 :
1066 6 : return ret;
1067 : }
1068 :
1069 2 : HcclResult PingMesh::HccnTarRemoveAttrInter(u32 targetNumInter, RpingInput *inputInter, PingTargetCommInfo *targetInter, std::shared_ptr<HcclSocket> &socketInter) {
1070 2 : HcclResult retInter = HCCL_SUCCESS;
1071 2 : u32 addressType = 0;
1072 2 : retInter = GetAddrType(&addressType);
1073 2 : if (retInter != HCCL_SUCCESS) {
1074 0 : HCCL_ERROR("[HccnTarRemoveAttrInter]GetAddrType Fail retInter %d", retInter);
1075 0 : return HCCL_E_PARA;
1076 : }
1077 4 : for (u32 i = 0; i < targetNumInter; i++) {
1078 : // 删除链路
1079 3 : std::unique_lock<std::mutex> lock(socketMapsMtx_);
1080 9 : if (socketMaps_.find(std::string(inputInter[i].dip.GetReadableIP())) == socketMaps_.end()) {
1081 1 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Socket[%s] doesn't exist.", inputInter[i].dip.GetReadableIP());
1082 1 : retInter = HCCL_E_NOT_FOUND;
1083 1 : break;
1084 : }
1085 4 : socketInter = socketMaps_[std::string(inputInter[i].dip.GetReadableIP())];
1086 2 : retInter = socketInter->DeInit();
1087 2 : if (retInter != HCCL_SUCCESS) {
1088 0 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Socket[%u][%s] deinit failed, ret[%d].", i, inputInter[i].dip.GetReadableIP(), retInter);
1089 0 : break;
1090 : }
1091 2 : lock.unlock();
1092 6 : if (!rdmaInfoMaps_.Find(std::string(inputInter[i].dip.GetReadableIP())).second) {
1093 0 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Target[%s] doesn't exist.", inputInter[i].dip.GetReadableIP());
1094 0 : retInter = HCCL_E_NOT_FOUND;
1095 0 : break;
1096 : }
1097 6 : if (!payloadLenMap_.Find(std::string(inputInter[i].dip.GetReadableIP())).second) {
1098 0 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Target[%s] doesn't exist.", inputInter[i].dip.GetReadableIP());
1099 0 : retInter = HCCL_E_NOT_FOUND;
1100 0 : break;
1101 : }
1102 4 : PingQpInfo *rdmainfo = &rdmaInfoMaps_[std::string(inputInter[i].dip.GetReadableIP())];
1103 2 : PingTargetInfo targetInfo { 0 };
1104 2 : if (addressType == HCCN_RPING_ADDR_TYPE_IP) {
1105 2 : retInter = RpingTargetAttrInit(targetInfo, inputInter[i], rdmainfo, false);
1106 : }
1107 2 : const char *socNamePtr = aclrtGetSocName();
1108 2 : CHK_PTR_NULL(socNamePtr);
1109 2 : if (addressType == HCCN_RPING_ADDR_TYPE_EID && IsSupportHCCLV2(socNamePtr)) {
1110 0 : retInter = RpingTargetAttrInitWithUb(targetInfo, inputInter[i], rdmainfo, false);
1111 : }
1112 :
1113 2 : targetInter[i] = targetInfo.remoteInfo;
1114 3 : }
1115 2 : return retInter;
1116 : }
1117 2 : HcclResult PingMesh::HccnRpingRemoveTarget(u32 deviceId, u32 targetNum, RpingInput *input)
1118 : {
1119 : // 判断当前状态
1120 2 : CHK_RET(RpingstateCheck(rpingState_, RpingState::READY));
1121 2 : CHK_RET(RpingstateCheck(rpingState_, RpingState::INITED)); // 所有目标都被移除时回到READY前的状态
1122 2 : if (pingHandle_ == nullptr) {
1123 0 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Device[%u] cannot add targets because it is not inited.", deviceId);
1124 0 : return HCCL_E_NOT_FOUND;
1125 : }
1126 2 : HCCL_INFO("[HccnRpingRemoveTarget]deviceId %u, targetNum %u", deviceId, targetNum);
1127 : // 调用hccp接口删除目标
1128 2 : HcclResult ret = HCCL_SUCCESS;
1129 2 : PingTargetCommInfo *target = new (std::nothrow) PingTargetCommInfo[targetNum];
1130 2 : CHK_PTR_NULL(target);
1131 2 : std::shared_ptr<HcclSocket> socket = nullptr;
1132 2 : ret = HccnTarRemoveAttrInter(targetNum, input, target, socket);
1133 2 : if (ret != HCCL_SUCCESS) {
1134 1 : delete[] target;
1135 1 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Target info is not correct, ret[%d].", ret);
1136 1 : return ret;
1137 : }
1138 1 : ret = hrtRaPingTargetDel(pingHandle_, target, targetNum);
1139 1 : delete[] target;
1140 1 : if (ret != HCCL_SUCCESS) {
1141 0 : HCCL_ERROR("[HCCN][HccnRpingRemoveTarget]Device[%u] remove targetNum %u failed, ret[%d].", deviceId, targetNum, ret);
1142 0 : return ret;
1143 : }
1144 1 : rpingTargetNum_ = rpingTargetNum_ - targetNum;
1145 1 : HCCL_INFO("[HCCN][HccnRpingRemoveTarget]Device[%u] remove targetNum %u success.", deviceId, targetNum);
1146 :
1147 : // 清除需要删掉的socket和rdma信息
1148 1 : std::unique_lock<std::mutex> lock(socketMapsMtx_);
1149 1 : RemoveMapInfo(input, targetNum, socketMaps_, rdmaInfoMaps_, payloadLenMap_);
1150 1 : lock.unlock();
1151 1 : if (rpingTargetNum_ <= 0) { // 目标数量小于等于0时, 记录的目标数量设为0, 切回初始化完成状态
1152 1 : rpingTargetNum_ = 0;
1153 1 : rpingState_ = RpingState::INITED;
1154 : }
1155 :
1156 1 : return HCCL_SUCCESS;
1157 2 : }
1158 :
1159 0 : HcclResult PingMesh::HccnRpingGetTarget(u32 deviceId, u32 targetNum, RpingInput *input, int *targetStat)
1160 : {
1161 0 : CHK_PTR_NULL(input);
1162 0 : CHK_PTR_NULL(targetStat);
1163 0 : std::unique_lock<std::mutex> lock(socketMapsMtx_);
1164 0 : for (u32 i = 0; i < targetNum; i++) {
1165 : //查询链路状态
1166 0 : if (socketMaps_.find(std::string(input[i].dip.GetReadableIP())) == socketMaps_.end()) {
1167 0 : HCCL_WARNING("[HCCN][HccnRpingGetTarget]Cannot get socket[%s]'s status.", input[i].dip.GetReadableIP());
1168 0 : targetStat[i] = static_cast<int>(RpingLinkState::DISCONNECTED);
1169 0 : continue;
1170 : }
1171 0 : HcclSocketStatus socketStatus = socketMaps_[std::string(input[i].dip.GetReadableIP())]->GetStatus();
1172 : // 转换状态信息
1173 0 : RpingLinkState linkStatus = ConvertHcclSocketStatus(socketStatus);
1174 : // 记录查询结果
1175 0 : targetStat[i] = static_cast<int>(linkStatus);
1176 : }
1177 0 : return HCCL_SUCCESS;
1178 0 : }
1179 :
1180 3 : HcclResult PingMesh::HccnRpingBatchPingStart(u32 deviceId, u32 pktNum, u32 interval, u32 timeout)
1181 : {
1182 : // 判断当前状态
1183 3 : CHK_RET(RpingstateCheck(rpingState_, RpingState::RUN));
1184 1 : HCCL_INFO("[HCCN][HccnRpingBatchPingStart]deviceId %u, pktNum %u, interval %u, timeout %u.", deviceId, pktNum,
1185 : interval, timeout);
1186 : // 调用hccp接口发起ping请求
1187 1 : if (pingHandle_ == nullptr) {
1188 0 : HCCL_ERROR("[HCCN][HccnRpingBatchPingStart]Device[%u] cannot start ping because it is not inited.", deviceId);
1189 0 : return HCCL_E_NOT_FOUND;
1190 : }
1191 : // 计算内存空间能否保存全部的payload信息，内存不足的话不可以发起ping请求
1192 1 : PingBufferInfo *bufferInfo = &(initInfo_.result);
1193 1 : u32 targetNum = rpingTargetNum_;
1194 1 : u32 payloadLen = pktNum * PING_TOTAL_PAYLOAD_MAX_SIZE * targetNum;
1195 1 : if ((bufferInfo->bufferSize != 0) && (payloadLen >= bufferInfo->bufferSize)) {
1196 0 : HCCL_ERROR("[HCCN][HccnRpingBatchPingStart]Buffer[%u] overflow threshold[%u], pktNum[%u], targetNum[%u].",
1197 : payloadLen, bufferInfo->bufferSize, pktNum, targetNum);
1198 0 : return HCCL_E_MEMORY;
1199 : }
1200 1 : PingTaskAttr attr = {};
1201 1 : attr.packetCnt = pktNum;
1202 1 : attr.packetInterval = interval;
1203 1 : attr.timeoutInterval = timeout;
1204 1 : CHK_RET(hrtRaPingTaskStart(pingHandle_, &attr));
1205 1 : HCCL_INFO("[HCCN][HccnRpingBatchPingStart]pingmesh task is started on device[%u].", deviceId);
1206 1 : rpingState_ = RpingState::RUN;
1207 1 : return HCCL_SUCCESS;
1208 : }
1209 :
1210 4 : HcclResult PingMesh::HccnRpingBatchPingStop(u32 deviceId)
1211 : {
1212 : // 判断当前状态
1213 4 : CHK_RET(RpingstateCheck(rpingState_, RpingState::STOP));
1214 1 : HCCL_INFO("[HCCN][HccnRpingBatchPingStop]deviceId %u", deviceId);
1215 : // 调用hccp接口中止ping请求
1216 1 : CHK_RET(hrtRaPingTaskStop(pingHandle_));
1217 1 : HCCL_INFO("[HCCN][HccnRpingBatchPingStop]Device[%u] pingmesh task is manually stopped.", deviceId);
1218 :
1219 1 : rpingState_ = RpingState::STOP;
1220 1 : return HCCL_SUCCESS;
1221 : }
1222 :
1223 2 : HcclResult PingMesh::HccnRpingGetResult(u32 deviceId, u32 targetNum, RpingInput *input, RpingOutput *output)
1224 : {
1225 2 : CHK_PTR_NULL(input);
1226 2 : CHK_PTR_NULL(output);
1227 2 : PingTargetResult *resultInfo = new (std::nothrow) PingTargetResult[targetNum];
1228 2 : CHK_PRT_RET(resultInfo == nullptr, HCCL_ERROR("[HCCN][HccnRpingGetResult]Alloc result memory failed."),
1229 : HCCL_E_MEMORY);
1230 2 : HCCL_INFO("[HCCN][HccnRpingGetResult]deviceId %u targetNum %u", deviceId, targetNum);
1231 2 : HcclResult ret = RpingResultInfoInit(resultInfo, rdmaInfoMaps_, input, targetNum);
1232 2 : if (ret != HCCL_SUCCESS) {
1233 0 : HCCL_ERROR("[HCCN][HccnRpingGetResult]RpingResultInfoInit failed,Device[%u] ret[%d] num[%u].", deviceId, ret, targetNum);
1234 0 : delete[] resultInfo;
1235 0 : return ret;
1236 : }
1237 : // 调用hccp接口获取探测结果
1238 2 : u32 num = targetNum; // resultinfo是一个带有返回值信息的数组, 对应的数组大小也需要返回，因此这里数组大小也传递指针
1239 2 : ret = hrtRaPingGetResults(pingHandle_, resultInfo, &num);
1240 2 : if (ret == HCCL_E_AGAIN) {
1241 0 : HCCL_WARNING("[HCCN][HccnRpingGetResult]Try again.");
1242 0 : delete[] resultInfo;
1243 0 : return HCCL_E_AGAIN;
1244 : }
1245 2 : if (ret != HCCL_SUCCESS || num > targetNum) {
1246 0 : HCCL_ERROR("[HCCN][HccnRpingGetResult]Device[%u] get result failed, ret[%d] num[%u], targetNum[%u].",
1247 : deviceId, ret, num, targetNum);
1248 0 : delete[] resultInfo;
1249 0 : return ret;
1250 : }
1251 2 : HCCL_INFO("[HCCN][HccnRpingGetResult]Device[%u] successfully gets [%u] results.", deviceId, num);
1252 :
1253 2 : GetResultFromReturnValue(resultInfo, output, num);
1254 :
1255 2 : delete[] resultInfo;
1256 2 : return HCCL_SUCCESS;
1257 : }
1258 :
1259 2 : HcclResult PingMesh::HccnRpingRefillPayloadHead(u8 *originalHead, u32 payloadNum)
1260 : {
1261 42 : for (u32 i = 0; i < payloadNum; i++) {
1262 40 : RpingIpHead *ipHead = reinterpret_cast<RpingIpHead*>(originalHead);
1263 : RpingIpHead ipHeadTmp;
1264 : // 清零之前记录头信息
1265 40 : errno_t memRet = memcpy_s(&ipHeadTmp, sizeof(RpingIpHead), ipHead, sizeof(RpingIpHead));
1266 40 : CHK_PRT_RET(memRet != EOK,
1267 : HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]copy head fail, ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1268 : memRet, &ipHeadTmp, sizeof(RpingIpHead), ipHead, sizeof(RpingIpHead)), HCCL_E_MEMORY);
1269 :
1270 : // 重填payload头
1271 40 : RpingPayloadHead *head = reinterpret_cast<RpingPayloadHead*>(originalHead);
1272 : // 清零要重填的内存
1273 40 : memRet = memset_s(originalHead, RPING_PAYLOAD_REFILL_LEN, 0, RPING_PAYLOAD_REFILL_LEN);
1274 40 : CHK_PRT_RET(memRet != EOK,
1275 : HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]clear first 136B fail, ret %d, destMaxSize %u, count %u",
1276 : memRet, RPING_PAYLOAD_REFILL_LEN, RPING_PAYLOAD_REFILL_LEN), HCCL_E_MEMORY);
1277 : // 清零payload头rsvd字段的内存
1278 40 : memRet = memset_s(head->reserved, RPING_PAYLOAD_RSVD_LEN, 0, RPING_PAYLOAD_RSVD_LEN);
1279 40 : CHK_PRT_RET(memRet != EOK,
1280 : HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]clear last 44B fail, ret %d, destMaxSize %u, count %u",
1281 : memRet, RPING_PAYLOAD_RSVD_LEN, RPING_PAYLOAD_RSVD_LEN), HCCL_E_MEMORY);
1282 : // 填充ip
1283 : HcclInAddr srcIpBinary;
1284 : HcclInAddr dstIpBinary;
1285 : // 报文来自对端，因此srcIp和dstIp需要调换过来
1286 40 : if (ipAddr_.GetFamily() == AF_INET) {
1287 20 : srcIpBinary.addr.s_addr = ipHeadTmp.ipv4.dstIp;
1288 20 : dstIpBinary.addr.s_addr = ipHeadTmp.ipv4.srcIp;
1289 : } else {
1290 20 : HcclInAddr *srcIpBinary6 = reinterpret_cast<HcclInAddr*>(ipHeadTmp.ipv6.srcIp);
1291 20 : HcclInAddr *dstIpBinary6 = reinterpret_cast<HcclInAddr*>(ipHeadTmp.ipv6.dstIp);
1292 20 : srcIpBinary = *dstIpBinary6;
1293 20 : dstIpBinary = *srcIpBinary6;
1294 : }
1295 40 : HcclIpAddress srcIp = HcclIpAddress(ipAddr_.GetFamily(), srcIpBinary);
1296 40 : u32 ipAddrStrLen = std::string(srcIp.GetReadableIP()).size();
1297 40 : memRet = memcpy_s(head->srcIp, IP_ADDRESS_BUFFER_LEN, srcIp.GetReadableIP(), ipAddrStrLen);
1298 40 : CHK_PRT_RET(memRet != EOK,
1299 : HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]Memcpy ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1300 : memRet, head->srcIp, IP_ADDRESS_BUFFER_LEN, srcIp.GetReadableIP(), ipAddrStrLen), HCCL_E_MEMORY);
1301 40 : HcclIpAddress dstIp = HcclIpAddress(ipAddr_.GetFamily(), dstIpBinary);
1302 40 : ipAddrStrLen = std::string(dstIp.GetReadableIP()).size();
1303 40 : memRet = memcpy_s(head->dstIp, IP_ADDRESS_BUFFER_LEN, dstIp.GetReadableIP(), ipAddrStrLen);
1304 40 : CHK_PRT_RET(memRet != EOK,
1305 : HCCL_ERROR("[HCCN][HccnRpingRefillPayloadHead]Memcpy ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1306 : memRet, head->dstIp, IP_ADDRESS_BUFFER_LEN, dstIp.GetReadableIP(), ipAddrStrLen), HCCL_E_MEMORY);
1307 : // 填充payloadLen
1308 120 : if (payloadLenMap_.Find(dstIp.GetReadableIP()).second) {
1309 0 : head->payloadLen = payloadLenMap_[dstIp.GetReadableIP()];
1310 : }
1311 : //填充addrtype
1312 40 : head->addrType = HCCN_RPING_ADDR_TYPE_IP;
1313 40 : originalHead += BYTE_PER_TARGET_DEFAULT;
1314 40 : }
1315 :
1316 2 : return HCCL_SUCCESS;
1317 : }
1318 :
1319 0 : HcclResult PingMesh::HccnRpingRefillUbPayloadHead(u8 *originalHead, u32 payloadNum)
1320 : {
1321 0 : for (u32 i = 0; i < payloadNum; i++) {
1322 0 : RpingEidHead *EidHead = reinterpret_cast<RpingEidHead*>(originalHead);
1323 : RpingEidHead EidHeadTmp;
1324 : // 清零之前记录头信息
1325 0 : errno_t memRet = memcpy_s(&EidHeadTmp, sizeof(RpingEidHead), EidHead, sizeof(RpingEidHead));
1326 0 : CHK_PRT_RET(memRet != EOK,
1327 : HCCL_ERROR("[HCCN][HccnRpingRefillUbPayloadHead]copy head fail, ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1328 : memRet, &EidHeadTmp, sizeof(RpingEidHead), EidHead, sizeof(RpingEidHead)), HCCL_E_MEMORY);
1329 :
1330 : // 重填payload头
1331 0 : RpingPayloadHead *head = reinterpret_cast<RpingPayloadHead*>(originalHead);
1332 : // 清零要重填的内存
1333 0 : memRet = memset_s(originalHead, RPING_PAYLOAD_UB_HEAD_LEN, 0, RPING_PAYLOAD_UB_HEAD_LEN);
1334 0 : CHK_PRT_RET(memRet != EOK,
1335 : HCCL_ERROR("[HCCN][HccnRpingRefillUbPayloadHead]clear first 256B fail, ret %d, destMaxSize %u, count %u",
1336 : memRet, RPING_PAYLOAD_UB_HEAD_LEN, RPING_PAYLOAD_UB_HEAD_LEN), HCCL_E_MEMORY);
1337 :
1338 : // 填充Eid
1339 0 : Eid srcEid;
1340 0 : Eid dstEid;
1341 : // 报文来自对端，因此srcEid和dstEid需要调换过来
1342 0 : Eid *srcEid6 = reinterpret_cast<Eid*>(EidHeadTmp.srcEid);
1343 0 : Eid *dstEid6 = reinterpret_cast<Eid*>(EidHeadTmp.dstEid);
1344 0 : srcEid = *dstEid6;
1345 0 : dstEid = *srcEid6;
1346 :
1347 0 : HcclIpAddress srcEidAddress = HcclIpAddress(srcEid);
1348 0 : u32 srcEidAddrStrLen = std::string(srcEidAddress.Describe()).size();
1349 0 : memRet = memcpy_s(head->srcEid, URMA_EID_LEN, srcEidAddress.Describe().c_str(), srcEidAddrStrLen);
1350 0 : CHK_PRT_RET(memRet != EOK,
1351 : HCCL_ERROR("[HCCN][HccnRpingRefillUbPayloadHead]Exchange eid fail. ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1352 : memRet, head->srcEid, URMA_EID_LEN, srcEidAddress.Describe().c_str(), srcEidAddrStrLen), HCCL_E_MEMORY);
1353 :
1354 0 : HcclIpAddress dstEidAddress = HcclIpAddress(dstEid);
1355 0 : u32 dstEidAddrStrLen = std::string(dstEidAddress.Describe()).size();
1356 0 : memRet = memcpy_s(head->dstEid, URMA_EID_LEN, dstEidAddress.Describe().c_str(), dstEidAddrStrLen);
1357 0 : CHK_PRT_RET(memRet != EOK,
1358 : HCCL_ERROR("[HCCN][HccnRpingRefillUbPayloadHead]Exchange eid fail. ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1359 : memRet, head->dstEid, URMA_EID_LEN, dstEidAddress.Describe().c_str(), dstEidAddrStrLen), HCCL_E_MEMORY);
1360 : // 填充payloadLen
1361 0 : if (payloadLenMap_.Find(dstEidAddress.Describe()).second) {
1362 0 : head->payloadLen = payloadLenMap_[dstEidAddress.Describe()];
1363 : }
1364 : //填充times
1365 0 : memRet = memcpy_s(head->timestamp, RPING_PAYLOAD_UB_TIME_LEN, EidHeadTmp.times, RPING_PAYLOAD_UB_TIME_LEN);
1366 0 : CHK_PRT_RET(memRet != EOK,
1367 : HCCL_ERROR("[HCCN][HccnRpingRefillUbPayloadHead]copy times fail. ret %d, dst:%p, dstMax:%u, src:%p, length:%u",
1368 : memRet, head->timestamp, RPING_PAYLOAD_UB_TIME_LEN, EidHeadTmp.times, RPING_PAYLOAD_UB_TIME_LEN), HCCL_E_MEMORY);
1369 : //填充taskID
1370 0 : head->rpingBatchId = EidHeadTmp.taskId;
1371 0 : head->addrType = HCCN_RPING_ADDR_TYPE_EID;
1372 0 : originalHead += BYTE_PER_TARGET_DEFAULT;
1373 0 : }
1374 0 : return HCCL_SUCCESS;
1375 : }
1376 :
1377 4 : HcclResult PingMesh::HccnRpingGetPayload(u32 deviceId, void **payload, u32 *payloadLen, HccnRpingMode mode)
1378 : {
1379 4 : CHK_PTR_NULL(payload);
1380 4 : CHK_PTR_NULL(payloadLen);
1381 : // 判断是否为payload配置过内存
1382 4 : if (!isUsePayload_) {
1383 2 : HCCL_DEBUG("[HCCN][HccnRpingGetPayload]not alloc memory on device[%u] for payload.", deviceId);
1384 2 : *payload = nullptr;
1385 2 : *payloadLen = 0;
1386 2 : return HCCL_SUCCESS;
1387 : }
1388 : // 将payload信息从device拷贝到host
1389 2 : PingBufferInfo *bufferInfo = &(initInfo_.result);
1390 2 : CHK_PRT_RET(bufferInfo->bufferSize == 0,
1391 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]no memory on device[%u] for payload.", deviceId), HCCL_E_MEMORY);
1392 : // payload_为空时，需要为其申请内存资源
1393 2 : if (payload_ == nullptr) {
1394 2 : payload_ = new (std::nothrow) u8[bufferInfo->bufferSize];
1395 2 : CHK_PRT_RET(payload_ == nullptr,
1396 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]Get payload from device[%u] failed.", deviceId), HCCL_E_MEMORY);
1397 : }
1398 : // 从device拷贝内存
1399 2 : HcclResult ret = HCCL_SUCCESS;
1400 2 : bool errorFlag = false;
1401 : do {
1402 4 : ret = hrtMemcpyEx(payload_, bufferInfo->bufferSize, reinterpret_cast<void *>(bufferInfo->bufferVa),
1403 2 : bufferInfo->bufferSize, rtMemcpyKind_t::RT_MEMCPY_DEVICE_TO_HOST);
1404 2 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
1405 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]Get payload from device[%u] failed, bufferSize[%u], bufferVa[%llu].", deviceId,
1406 : bufferInfo->bufferSize, bufferInfo->bufferVa), (errorFlag = true));
1407 : // 重填payload头
1408 2 : u32 payloadNum = bufferInfo->bufferSize / BYTE_PER_TARGET_DEFAULT;
1409 2 : u8 *payloadTmp = payload_;
1410 2 : if (mode == HCCN_RPING_MODE_ROCE) {
1411 2 : ret = HccnRpingRefillPayloadHead(payloadTmp, payloadNum);
1412 2 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
1413 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]HccnRpingRefillPayloadHead failed."), errorFlag = true);
1414 : }
1415 2 : const char *socNamePtr = aclrtGetSocName();
1416 2 : CHK_PRT_BREAK(socNamePtr == nullptr,
1417 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]aclrtGetSocName failed."), errorFlag = true);
1418 2 : if (mode == HCCN_RPING_MODE_UB && IsSupportHCCLV2(socNamePtr)) {
1419 0 : ret = HccnRpingRefillUbPayloadHead(payloadTmp, payloadNum);
1420 0 : CHK_PRT_BREAK(ret != HCCL_SUCCESS,
1421 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]HccnRpingRefillUbPayloadHead failed."), errorFlag = true);
1422 : }
1423 2 : *payload = payload_;
1424 2 : *payloadLen = bufferInfo->bufferSize;
1425 : } while (0);
1426 :
1427 2 : if (errorFlag) {
1428 0 : HCCL_ERROR("[HCCN][HccnRpingGetPayload]Get payload from device[%u] failed, bufferSize[%u], bufferVa[%llu].", deviceId,
1429 : bufferInfo->bufferSize, bufferInfo->bufferVa);
1430 0 : delete[] payload_;
1431 0 : payload_ = nullptr;
1432 0 : return ret;
1433 : }
1434 2 : return HCCL_SUCCESS;
1435 : }
1436 :
1437 : }
|
