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