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