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