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 : #include "orion_adapter_hccp.h"
11 : #include <chrono>
12 : #include <unistd.h>
13 : #include <memory>
14 : #include <unordered_map>
15 : #include "sal.h"
16 : #include "network_api_exception.h"
17 : #include "internal_exception.h"
18 : #include "hccp.h"
19 : #include "hccp_tlv.h"
20 : #include "hccp_ctx.h"
21 : #include "hccp_async.h"
22 : #include "env_config_v2.h"
23 : #include "hccp_common.h"
24 : #include "exception_util.h"
25 : #include "adapter_error_manager_pub.h"
26 :
27 : using namespace std;
28 :
29 : namespace Hccl {
30 : constexpr u32 ONE_HUNDRED_MICROSECOND_OF_USLEEP = 100;
31 : constexpr u32 ONE_MILLISECOND_OF_USLEEP = 1000;
32 : constexpr unsigned int SOCKET_NUM_ONE = 1;
33 : constexpr u32 MAX_NUM_OF_WHITE_LIST_NUM = 16;
34 : constexpr u32 AUTO_LISTEN_PORT = 0;
35 : constexpr u64 SOCKET_SEND_MAX_SIZE = 0x7FFFFFFFFFFFFFFF;
36 : constexpr u32 MAX_WR_NUM = 1024;
37 : constexpr u32 MAX_SEND_SGE_NUM = 1;
38 : constexpr u32 MAX_RECV_SGE_NUM = 1;
39 : constexpr u32 MAX_CQ_DEPTH = 65535;
40 : constexpr u32 NDA_CQ_DEPTH_FOR_UBNIC = 31 * 1024;
41 : constexpr u32 NDA_CQ_DEPTH_FOR_XSCDV = 32 * 1024;
42 : constexpr u32 MAX_INLINE_DATA = 64;
43 : constexpr u32 RA_TLV_REQUEST_UNAVAIL = 128308;
44 : constexpr u32 ROCE_ENOMEM_RET = 328100;
45 : constexpr u32 GET_TP_ATTR_OPCODE = 106;
46 : constexpr u32 GET_TLS_ENABLE_OPCODE = 95;
47 : constexpr u32 GET_TLS_ENABLE_VERSION = 1;
48 : constexpr u32 GET_TP_ATTR_VERSION = 2;
49 :
50 : const std::unordered_map<HrtNetworkMode, NetworkMode, EnumClassHash> HRT_NETWORK_MODE_MAP
51 : = {{HrtNetworkMode::PEER, NetworkMode::NETWORK_PEER_ONLINE}, {HrtNetworkMode::HDC, NetworkMode::NETWORK_OFFLINE}};
52 :
53 : s32 g_linkTimeout = 0;
54 41 : inline s32 EnvLinkTimeoutGet()
55 : {
56 41 : g_linkTimeout = g_linkTimeout != 0 ? g_linkTimeout : EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut();
57 41 : return g_linkTimeout;
58 : }
59 :
60 6 : HcclResult HrtRaGetTlsStatus(struct RaInfo* info, TlsStatus& tlsStatus)
61 : {
62 6 : tlsStatus = TlsStatus::UNKNOWN;
63 9 : CHK_PTR_NULL(info);
64 :
65 5 : u32 tlsVersion = 0;
66 5 : s32 versionRet = RaGetInterfaceVersion(info->phyId, GET_TLS_ENABLE_OPCODE, &tlsVersion);
67 5 : if (versionRet != 0 || tlsVersion < GET_TLS_ENABLE_VERSION) {
68 6 : HCCL_WARNING(
69 : "[HrtRaGetTlsStatus] this package does not support RaGetTlsEnable for device, "
70 : "please change new package. ret[%d], tlsVersion[%u].",
71 : versionRet, tlsVersion);
72 2 : return HCCL_E_NOT_SUPPORT;
73 : }
74 :
75 3 : bool tlsEnable = false;
76 3 : s32 ret = RaGetTlsEnable(info, &tlsEnable);
77 3 : if (ret != 0) {
78 1 : tlsStatus = TlsStatus::DISABLE;
79 3 : HCCL_ERROR(
80 : "[HrtRaGetTlsStatus] errNo[0x%016llx] failed ret[%d], phyId[%u]", HCCL_ERROR_CODE(HCCL_E_NETWORK), ret,
81 : info->phyId);
82 1 : return HCCL_E_NETWORK;
83 : }
84 :
85 2 : tlsStatus = tlsEnable ? TlsStatus::ENABLE : TlsStatus::DISABLE;
86 6 : HCCL_INFO(
87 : "[HrtRaGetTlsStatus] phyId[%u], tlsEnable[%d], tlsStatus[%d]", info->phyId, tlsEnable,
88 : static_cast<s32>(tlsStatus));
89 2 : return HCCL_SUCCESS;
90 : }
91 :
92 89 : inline union HccpIpAddr IpAddressToHccpIpAddr(IpAddress& addr)
93 : {
94 : union HccpIpAddr hccpIpAddr;
95 89 : if (addr.GetFamily() == AF_INET) {
96 89 : hccpIpAddr.addr = addr.GetBinaryAddress().addr;
97 : } else {
98 0 : hccpIpAddr.addr6 = addr.GetBinaryAddress().addr6;
99 : }
100 89 : return hccpIpAddr;
101 : }
102 :
103 2 : inline IpAddress IfAddrInfoToIpAddress(struct InterfaceInfo info)
104 : {
105 : BinaryAddr addr;
106 2 : if (info.family == AF_INET) {
107 2 : addr.addr = info.ifaddr.ip.addr;
108 : } else {
109 0 : addr.addr6 = info.ifaddr.ip.addr6;
110 : }
111 4 : return IpAddress(addr, info.family, info.scopeId);
112 : }
113 :
114 6 : void* HrtRaTlvInit(HRaTlvInitConfig& cfg)
115 : {
116 18 : HCCL_INFO("[Init][RaTlv] Input params: version=[%d], phyId=[%u], mode=[%u]", cfg.version, cfg.phyId, cfg.mode);
117 6 : struct TlvInitInfo init_info {};
118 6 : init_info.version = cfg.version;
119 6 : init_info.phyId = cfg.phyId;
120 6 : init_info.nicPosition = HRT_NETWORK_MODE_MAP.at(cfg.mode);
121 :
122 6 : s32 ret = 0;
123 : unsigned int buffer_size;
124 : void* tlv_handle;
125 :
126 6 : ret = RaTlvInit(&init_info, &buffer_size, &tlv_handle);
127 6 : if (ret != 0 || tlv_handle == nullptr) {
128 4 : MACRO_THROW(
129 : NetworkApiException, StringFormat(
130 : "[Init][RaTlv]errNo[0x%016llx] ra tlv init fail. params: mode=%u, device id=%u, "
131 : "version=%d, tlv_handle=%p, return: ret[%d]",
132 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), cfg.mode, init_info.phyId,
133 : init_info.version, tlv_handle, ret));
134 : }
135 :
136 15 : HCCL_INFO("tlv init success, device id[%u]", init_info.phyId);
137 :
138 5 : return tlv_handle;
139 : }
140 :
141 4 : HcclResult HrtRaTlvRequest(void* tlv_handle, u32 tlv_module_type, u32 tlv_ccu_msg_type)
142 : {
143 4 : CHK_PTR_NULL(tlv_handle);
144 :
145 12 : HCCL_INFO(
146 : "[Request][RaTlv] Input params: tlv_handle=[%p], tlv_module_type=[%u], tlv_ccu_msg_type=[%u]", tlv_handle,
147 : tlv_module_type, tlv_ccu_msg_type);
148 4 : s32 ret = 0;
149 :
150 4 : struct TlvMsg send_msg {};
151 4 : struct TlvMsg recv_msg {};
152 4 : send_msg.type = tlv_ccu_msg_type;
153 :
154 4 : ret = RaTlvRequest(tlv_handle, tlv_module_type, &send_msg, &recv_msg);
155 4 : if (ret != 0) {
156 1 : if (ret == RA_TLV_REQUEST_UNAVAIL || ret == OTHERS_ENOTSUPP) {
157 0 : HCCL_WARNING("[HrtRaTlvRequest]ra tlv request UNAVAIL. return: ret[%d]", ret);
158 0 : return HCCL_E_UNAVAIL;
159 : }
160 4 : MACRO_THROW(
161 : NetworkApiException,
162 : StringFormat(
163 : "[Request][RaTlv]errNo[0x%016llx] ra tlv request fail. params: tlv_handle=%p, tlv_module_type=%u, "
164 : "tlv_ccu_msg_type=%u, return: ret=%d",
165 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), tlv_handle, tlv_module_type, tlv_ccu_msg_type, ret));
166 : }
167 :
168 9 : HCCL_INFO("tlv request success, tlv module type[%u], message type[%u]", tlv_module_type, tlv_ccu_msg_type);
169 3 : return HCCL_SUCCESS;
170 : }
171 :
172 4147 : void HrtRaTlvRequestForCustomChannel(void* tlvHandle, u32 msgType, void* customIn, void* customOut)
173 : {
174 8294 : CHECK_NULLPTR(tlvHandle, "[HrtRaTlvRequestForCustomChannel] tlvHandle is nullptr!");
175 8292 : CHECK_NULLPTR(customIn, "[HrtRaTlvRequestForCustomChannel] customIn is nullptr!");
176 4146 : CHECK_NULLPTR(customOut, "[HrtRaTlvRequestForCustomChannel] customOut is nullptr!");
177 :
178 4144 : struct TlvMsg sendMsg {};
179 4144 : sendMsg.type = msgType;
180 4144 : sendMsg.length = sizeof(CustomChanInfoIn);
181 4144 : sendMsg.data = static_cast<char*>(customIn);
182 :
183 4144 : struct TlvMsg recvMsg {};
184 4144 : recvMsg.type = msgType;
185 4144 : recvMsg.length = sizeof(CustomChanInfoOut);
186 4144 : recvMsg.data = static_cast<char*>(customOut);
187 :
188 4144 : s32 ret = RaTlvRequest(tlvHandle, TLV_MODULE_TYPE_CCU, &sendMsg, &recvMsg);
189 4144 : if (ret != 0) {
190 4 : MACRO_THROW(NetworkApiException, StringFormat("[%s] RaTlvRequest fail, ret[%d]", __func__, ret));
191 : }
192 4143 : }
193 :
194 6 : void HrtRaTlvDeInit(void* tlv_handle)
195 : {
196 6 : CHECK_NULLPTR(tlv_handle, "[HrtRaTlvDeInit] tlv_handle is nullptr!");
197 :
198 6 : s32 ret = 0;
199 :
200 6 : ret = RaTlvDeinit(tlv_handle);
201 6 : if (ret != 0) {
202 4 : MACRO_THROW(
203 : NetworkApiException,
204 : StringFormat(
205 : "[DeInit][RaTlv]errNo[0x%016llx] ra tlv deinit fail. params: tlv_handle=%p, return: ret=%d",
206 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), tlv_handle, ret));
207 : }
208 5 : }
209 :
210 11 : void HrtRaInit(HRaInitConfig& cfg)
211 : {
212 33 : HCCL_INFO("[Init][Ra] Input params: phyId=[%u], mode=[%u]", cfg.phyId, cfg.mode);
213 :
214 11 : struct RaInitConfig config {};
215 11 : config.phyId = cfg.phyId;
216 11 : config.nicPosition = HRT_NETWORK_MODE_MAP.at(cfg.mode);
217 11 : config.hdcType = PID_HDC_TYPE;
218 11 : config.enableHdcAsync = true;
219 :
220 11 : s32 ret = 0;
221 11 : auto startTime = std::chrono::steady_clock::now();
222 11 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
223 :
224 : while (true) {
225 11 : ret = RaInit(&config);
226 11 : if (!ret) {
227 10 : break; // 成功跳出
228 1 : } else if (ret == SOCK_EAGAIN) {
229 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
230 0 : if (bTimeout) {
231 0 : MACRO_THROW(
232 : NetworkApiException,
233 : StringFormat(
234 : "[Init][Ra]errNo[0x%016llx], ra init timeout[%lld s], phy_id=%u, nic_position=%u, ret=%d",
235 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), timeout, config.phyId, config.nicPosition, ret));
236 : }
237 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
238 : } else {
239 : // 非ra限速场景错误，不轮询。直接退出
240 4 : MACRO_THROW(
241 : NetworkApiException,
242 : StringFormat(
243 : "[Init][Ra]errNo[0x%016llx] ra init fail, phy_id=%u, nic_position=%u, ret=%d",
244 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), config.phyId, config.nicPosition, ret));
245 : }
246 0 : }
247 30 : HCCL_INFO("init ra success,return: ret[%d]", ret);
248 10 : }
249 :
250 4 : void HrtRaDeInit(HRaInitConfig& cfg)
251 : {
252 12 : HCCL_INFO("[DeInit][Ra] Input params: phyId=[%u], mode=[%u]", cfg.phyId, cfg.mode);
253 4 : struct RaInitConfig config {};
254 4 : config.phyId = cfg.phyId;
255 4 : config.nicPosition = HRT_NETWORK_MODE_MAP.at(cfg.mode);
256 4 : config.hdcType = PID_HDC_TYPE;
257 :
258 4 : s32 ret = 0;
259 4 : auto startTime = std::chrono::steady_clock::now();
260 4 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
261 : while (true) {
262 4 : ret = RaDeinit(&config);
263 4 : if (!ret) {
264 9 : HCCL_INFO("deinit ra success,return: ret[%d]", ret);
265 3 : break; // 成功跳出
266 1 : } else if (ret == SOCK_EAGAIN) {
267 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
268 0 : if (bTimeout) {
269 0 : MACRO_THROW(
270 : NetworkApiException,
271 : StringFormat(
272 : "[DeInit][Ra]errNo[0x%016llx] ra deinit timeout[%lld s], phy_id=%u, nic_position=%u, ret=%d",
273 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), timeout, config.phyId, config.nicPosition, ret));
274 : }
275 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
276 : } else {
277 : // 非ra限速场景错误，不轮询。直接退出
278 4 : MACRO_THROW(
279 : NetworkApiException,
280 : StringFormat(
281 : "[DeInit][Ra]errNo[0x%016llx] ra deinit fail, phy_id=%u, nic_position=%u, ret=%d",
282 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), config.phyId, config.nicPosition, ret));
283 : }
284 0 : }
285 3 : }
286 :
287 0 : static void SocketBatchConnect(SocketConnectInfoT conn[], u32 num)
288 : {
289 0 : CHECK_NULLPTR(conn, "[SocketBatchConnect] conn is nullptr!");
290 0 : HCCL_INFO("[BatchConnect][RaSocket] Input params: num=%u", num);
291 0 : s32 ret = 0;
292 0 : auto startTime = std::chrono::steady_clock::now();
293 0 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
294 : while (true) {
295 0 : ret = RaSocketBatchConnect(conn, num);
296 0 : if (!ret) {
297 0 : HCCL_INFO("socket batch connect success, ret=%d", ret);
298 0 : break; // 成功跳出
299 0 : } else if (ret == SOCK_EAGAIN) {
300 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
301 0 : if (bTimeout) {
302 0 : MACRO_THROW(
303 : NetworkApiException, StringFormat(
304 : "[BatchConnect][RaSocket]errNo[0x%016llx] ra socket batch connect, "
305 : "timeout[%lld s]. return[%d], params: num[%u]",
306 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), timeout, ret, num));
307 : }
308 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
309 : } else {
310 0 : MACRO_THROW(
311 : NetworkApiException, StringFormat(
312 : "[BatchConnect][RaSocket]errNo[0x%016llx] ra socket batch connect fail, "
313 : "return[%d], params: num[%u]",
314 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret, num));
315 : }
316 0 : }
317 0 : }
318 :
319 0 : void HrtRaSocketConnectOne(RaSocketConnectParam& in)
320 : {
321 0 : HCCL_INFO(
322 : "[ConnectOne][RaSocket] Input params: socketHandle=%p, remoteIp=%s, port=%u, tag=%s", in.socketHandle,
323 : in.remoteIp.Describe().c_str(), in.port, in.tag.c_str());
324 :
325 0 : struct SocketConnectInfoT connInfo {};
326 0 : connInfo.socketHandle = in.socketHandle;
327 0 : connInfo.remoteIp = IpAddressToHccpIpAddr(in.remoteIp);
328 0 : connInfo.port = in.port;
329 :
330 0 : int sret = strcpy_s(connInfo.tag, sizeof(connInfo.tag), in.tag.c_str());
331 0 : if (sret != 0) {
332 : string msg = StringFormat(
333 : "[HrtRaSocketConnectOne] copy tag[%s] to hccp tag failed, in.tag size[%d], connInfo.tag size[%d], ret[%d]",
334 0 : in.tag.c_str(), sizeof(in.tag.c_str()), sizeof(connInfo.tag), sret);
335 0 : MACRO_THROW(NetworkApiException, msg);
336 0 : }
337 :
338 0 : HCCL_INFO("Socket Connect tag=[%s], remoteIp[%s]", connInfo.tag, in.remoteIp.Describe().c_str());
339 0 : SocketBatchConnect(&connInfo, 1);
340 0 : }
341 :
342 11 : static void HRaSocketBatchClose(struct SocketCloseInfoT conn[], u32 num)
343 : {
344 11 : CHECK_NULLPTR(conn, "[HRaSocketBatchClose] conn is nullptr!");
345 33 : HCCL_INFO("[BatchClose][RaSocket] Input params: num=%u", num);
346 33 : HCCL_INFO("ra socket batch close");
347 11 : s32 ret = 0;
348 11 : auto startTime = std::chrono::steady_clock::now();
349 11 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
350 : while (true) {
351 11 : ret = RaSocketBatchClose(conn, num);
352 11 : if (!ret) {
353 33 : HCCL_INFO("socket batch close success, ret=%d", ret);
354 11 : break; // 成功跳出
355 0 : } else if (ret == SOCK_EAGAIN) {
356 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
357 0 : if (bTimeout) {
358 0 : MACRO_THROW(
359 : NetworkApiException, StringFormat(
360 : "[BatchClose][RaSocket]errNo[0x%016llx] ra socket batch close, timeout[%d "
361 : "s], return[%d], params: num[%u]",
362 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), timeout, ret, num));
363 : }
364 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
365 : } else {
366 : // 非ra限速场景错误，不轮询，直接退出
367 0 : MACRO_THROW(
368 : NetworkApiException,
369 : StringFormat(
370 : "[BatchClose][RaSocket]errNo[0x%016llx] ra socket batch close fail, return[%d], params: num[%u]",
371 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret, num));
372 : }
373 0 : }
374 11 : }
375 :
376 11 : void HrtRaSocketCloseOne(RaSocketCloseParam& in)
377 : {
378 33 : HCCL_INFO("[CloseOne][RaSocket] Input params: socketHandle=%p, fdHandle=%p", in.socketHandle, in.fdHandle);
379 11 : struct SocketCloseInfoT closeInfo = {};
380 11 : closeInfo.fdHandle = in.fdHandle;
381 11 : closeInfo.socketHandle = in.socketHandle;
382 :
383 11 : HRaSocketBatchClose(&closeInfo, 1);
384 11 : }
385 :
386 0 : static void ReportAddrInUseError(const IpAddress& localIp, u32 port, HrtNetworkMode netMode)
387 : {
388 0 : std::string errMsg = "The IP address " + std::string(localIp.Describe().c_str()) + " and port "
389 0 : + std::to_string(port) + " have already been bound.";
390 0 : if (netMode == HrtNetworkMode::PEER) {
391 0 : RPT_INPUT_ERR(true, "EI0019", std::vector<std::string>({"reason"}), std::vector<std::string>({errMsg}));
392 : } else {
393 0 : RPT_INPUT_ERR(true, "EI0020", std::vector<std::string>({"reason"}), std::vector<std::string>({errMsg}));
394 : }
395 0 : }
396 :
397 : static void
398 1 : HRaSocketListenStart(struct SocketListenInfoT conn[], u32 num, const IpAddress& localIp, HrtNetworkMode netMode)
399 : {
400 1 : CHECK_NULLPTR(conn, "[HRaSocketListenStart] conn is nullptr!");
401 3 : HCCL_INFO("[ListenStart][RaSocket] Input params: num=%u", num);
402 1 : s32 ret = 0;
403 1 : auto startTime = std::chrono::steady_clock::now();
404 1 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
405 :
406 : while (true) {
407 949 : ret = RaSocketListenStart(conn, num);
408 949 : if (ret == 0) {
409 0 : HCCL_INFO("socket listen start success, ret=%d", ret);
410 0 : break;
411 949 : } else if (ret == SOCK_EAGAIN) {
412 949 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
413 949 : if (bTimeout) {
414 4 : MACRO_THROW(
415 : NetworkApiException,
416 : StringFormat(
417 : "[ListenStart][RaSocket]errNo[0x%016llx] ra socket listen start, timeout[%d s], return[%d], "
418 : "params: num[%u]",
419 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), EnvLinkTimeoutGet(), ret, num));
420 : }
421 948 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
422 0 : } else if (ret == SOCK_EADDRINUSE) {
423 0 : u32 port = (num > 0) ? conn[0].port : HCCL_INVALID_PORT;
424 0 : ReportAddrInUseError(localIp, port, netMode);
425 0 : MACRO_THROW(
426 : NetworkApiException,
427 : StringFormat(
428 : "[%s]ra socket listen could not start, due to the port[%u] has already been bound. please try"
429 : " another port or check the port status",
430 : __func__, port));
431 0 : } else if (ret == SOCK_EADDRNOTAVAIL) {
432 0 : MACRO_THROW(
433 : NetworkApiException,
434 : StringFormat(
435 : "[%s] Socket listen start fail: "
436 : "IP address is not available, please check the IP address configuration, return[%d]",
437 : __func__, ret));
438 : } else {
439 : // 非ra限速场景错误，不轮询，直接退出
440 0 : MACRO_THROW(
441 : NetworkApiException,
442 : StringFormat(
443 : "[ListenStart][RaSocket]errNo[0x%016llx] ra socket listen start fail, return[%d], params: num[%u]",
444 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), EnvLinkTimeoutGet(), ret, num));
445 : }
446 948 : }
447 0 : }
448 :
449 : static bool
450 3 : RaSocketTryListenStart(struct SocketListenInfoT conn[], u32 num, const IpAddress& localIp, HrtNetworkMode netMode)
451 : {
452 3 : CHECK_NULLPTR(conn, "[RaSocketTryListenStart] conn is nullptr!");
453 9 : HCCL_INFO("[TryListenStart][RaSocket] Input params: num=%u", num);
454 3 : s32 ret = RaSocketListenStart(conn, num);
455 3 : if (ret == 0) {
456 2 : return true;
457 1 : } else if (ret == SOCK_EAGAIN) {
458 0 : HCCL_INFO("[%s] listen eagain", __func__);
459 0 : return true;
460 1 : } else if (ret == SOCK_EADDRINUSE) {
461 0 : u32 port = (num > 0) ? conn[0].port : HCCL_INVALID_PORT;
462 0 : HCCL_INFO(
463 : "[%s]ra socket listen could not start, due to the port[%u] has already been bound. please try"
464 : " another port or check the port status",
465 : __func__, port);
466 0 : return false;
467 1 : } else if (ret == SOCK_EADDRNOTAVAIL) {
468 4 : MACRO_THROW(
469 : NetworkApiException,
470 : StringFormat(
471 : "[%s] Socket listen start fail: "
472 : "IP address is not available, please check the IP address configuration, return[%d]",
473 : __func__, ret));
474 : } else {
475 : // 非ra限速场景错误，不轮询，直接退出
476 0 : MACRO_THROW(
477 : NetworkApiException,
478 : StringFormat(
479 : "[TryListenStart][RaSocket]errNo[0x%016llx] ra socket listen start fail, return[%d], params: num[%u]",
480 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret, num));
481 : }
482 : }
483 :
484 2 : static void HRaSocketListenStop(struct SocketListenInfoT conn[], u32 num)
485 : {
486 2 : CHECK_NULLPTR(conn, "[HRaSocketListenStop] conn is nullptr!");
487 6 : HCCL_INFO("[ListenStop][RaSocket] Input params: num=%u", num);
488 2 : s32 ret = 0;
489 2 : auto startTime = std::chrono::steady_clock::now();
490 2 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
491 : while (true) {
492 2 : ret = RaSocketListenStop(conn, num);
493 2 : if (!ret || ret == 228202) { // 待修改: 同步版本后 228202 修改为 SOCK_ENODEV
494 6 : HCCL_INFO("socket listen stop success, ret=%d", ret);
495 2 : break; // 成功跳出
496 0 : } else if (ret == SOCK_EAGAIN) {
497 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
498 0 : if (bTimeout) {
499 0 : MACRO_THROW(
500 : NetworkApiException, StringFormat(
501 : "[ListenStop][RaSocket]errNo[0x%016llx] ra socket listen stop fail, "
502 : "timeout[%d s], return[%d], params: num[%u]",
503 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), timeout, ret, num));
504 : }
505 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
506 : } else {
507 : // 非ra限速场景错误，不轮询，直接退出
508 0 : MACRO_THROW(
509 : NetworkApiException,
510 : StringFormat(
511 : "[ListenStop][RaSocket]errNo[0x%016llx] ra socket listen stop fail, return[%d], params: num[%u]",
512 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret, num));
513 : }
514 0 : }
515 2 : }
516 :
517 1 : void HrtRaSocketListenOneStart(RaSocketListenParam& in, HrtNetworkMode netMode)
518 : {
519 3 : HCCL_INFO("[ListenStart][RaSocket] Input params: socketHandle: %p, port: %u", in.socketHandle, in.port);
520 1 : struct SocketListenInfoT listenInfo {};
521 1 : listenInfo.socketHandle = in.socketHandle;
522 1 : listenInfo.port = in.port;
523 1 : HRaSocketListenStart(&listenInfo, 1, in.localIp, netMode);
524 0 : }
525 :
526 3 : bool HrtRaSocketTryListenOneStart(RaSocketListenParam& in, HrtNetworkMode netMode)
527 : {
528 9 : HCCL_INFO("[TryListenOneStart][RaSocket] Input params: socketHandle: %p, port: %u", in.socketHandle, in.port);
529 3 : struct SocketListenInfoT listenInfo {};
530 3 : listenInfo.socketHandle = in.socketHandle;
531 3 : listenInfo.port = in.port;
532 3 : bool ret = RaSocketTryListenStart(&listenInfo, 1, in.localIp, netMode);
533 2 : if (ret && in.port == AUTO_LISTEN_PORT) {
534 0 : in.port = listenInfo.port;
535 : }
536 2 : return ret;
537 : }
538 :
539 2 : void HrtRaSocketListenOneStop(RaSocketListenParam& in)
540 : {
541 6 : HCCL_INFO("[ListenOneStop][RaSocket] Input params: socketHandle: %p, port: %u", in.socketHandle, in.port);
542 2 : struct SocketListenInfoT listenInfo {};
543 2 : listenInfo.socketHandle = in.socketHandle;
544 2 : listenInfo.port = in.port;
545 2 : HRaSocketListenStop(&listenInfo, 1);
546 2 : }
547 :
548 1 : void RaBlockGetSockets(u32 role, SocketInfoT conn[], u32 num, u32 timeoutSec) // 修改为内部函数，不对外
549 : {
550 1 : CHECK_NULLPTR(conn, "[RaBlockGetSockets] conn is nullptr!");
551 3 : HCCL_INFO("[GetSockets][RaBlock] Input params: role=[%u], num=[%u], timeoutSec=[%u s]", role, num, timeoutSec);
552 : s32 sockRet;
553 1 : u32 gotSocketsCnt = 0;
554 1 : auto startTime = std::chrono::steady_clock::now();
555 1 : auto linkTimeout = std::chrono::seconds(EnvLinkTimeoutGet());
556 : auto timeout
557 1 : = (timeoutSec > 0 && timeoutSec < linkTimeout.count()) ? std::chrono::seconds(timeoutSec) : linkTimeout;
558 : while (true) {
559 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
560 0 : MACRO_THROW(
561 : NetworkApiException,
562 : StringFormat(
563 : "[HrtRaBlockGetSockets] get rasocket timeout role[%u], num[%u], gotSocketsCnt[%u], timeout[%lld]s",
564 : role, num, gotSocketsCnt, timeout));
565 : }
566 1 : u32 connectedNum = 0;
567 1 : sockRet = RaGetSockets(role, conn, num, &connectedNum);
568 1 : if ((connectedNum == 0 && sockRet == 0) || (sockRet == SOCK_EAGAIN)) {
569 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
570 1 : } else if (sockRet != 0) {
571 0 : MACRO_THROW(
572 : NetworkApiException,
573 : StringFormat(
574 : "[Get][RaSocket]get rasocket error. role[%u], num[%u], sockRet[%d], connectednum[%u]", role, num,
575 : sockRet, connectedNum));
576 : } else {
577 1 : gotSocketsCnt += connectedNum;
578 1 : if (gotSocketsCnt == num) {
579 3 : HCCL_INFO("block get sockets success, socket num[%u]", gotSocketsCnt);
580 1 : break;
581 0 : } else if (gotSocketsCnt > num) {
582 0 : MACRO_THROW(
583 : NetworkApiException,
584 : StringFormat("[Get][RaSocket]total Sockets[%u], more than needed num[%u]!", gotSocketsCnt, num));
585 : } else {
586 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
587 : }
588 : }
589 0 : }
590 1 : }
591 :
592 1 : RaSocketFdHandleParam HrtRaBlockGetOneSocket(u32 role, RaSocketGetParam& param, u32 timeout)
593 : {
594 3 : HCCL_INFO(
595 : "[GetOneSocket][RaSocket] Input params: role=[%u],socketHandle=[%p], fdHandle=[%p], remoteIp=[%s], timeout=[%u "
596 : "s]",
597 : role, param.socketHandle, param.fdHandle, param.remoteIp.Describe().c_str(), timeout);
598 1 : struct SocketInfoT socketInfo {};
599 :
600 1 : socketInfo.socketHandle = param.socketHandle;
601 1 : socketInfo.fdHandle = param.fdHandle;
602 1 : socketInfo.remoteIp = IpAddressToHccpIpAddr(param.remoteIp);
603 1 : socketInfo.status = SOCKET_NOT_CONNECTED;
604 :
605 1 : int sret = strcpy_s(socketInfo.tag, sizeof(socketInfo.tag), param.tag.c_str());
606 1 : if (sret != 0) {
607 0 : MACRO_THROW(
608 : NetworkApiException,
609 : StringFormat(
610 : "[HrtRaBlockGetOneSocket] copy tag[%s] to hccp failed, ret=%d, role=%u,socketHandle=%p, fdHandle=%p, "
611 : "remoteIp=%s, socketInfo.tag size=%d, param.tag size=%d",
612 : param.tag.c_str(), sret, role, param.socketHandle, param.fdHandle, param.remoteIp.Describe().c_str(),
613 : sizeof(socketInfo.tag), sizeof(param.tag.c_str())));
614 : }
615 :
616 3 : HCCL_INFO("Socket Get tag=[%s], remoteIp[%s], ret[%d]", socketInfo.tag, param.remoteIp.Describe().c_str(), sret);
617 1 : RaBlockGetSockets(role, &socketInfo, 1, timeout);
618 :
619 1 : return RaSocketFdHandleParam(socketInfo.fdHandle, socketInfo.status);
620 : }
621 :
622 0 : void HrtRaSocketBlockSend(const FdHandle fdHandle, const void* data, u32 sendSize)
623 : {
624 0 : CHECK_NULLPTR(fdHandle, "[HrtRaSocketBlockSend] fdHandle is nullptr!");
625 0 : CHECK_NULLPTR(data, "[HrtRaSocketBlockSend] data is nullptr!");
626 0 : s32 ret = 0;
627 0 : void* sendData = const_cast<void*>(data);
628 0 : const std::chrono::seconds timeout = std::chrono::seconds(EnvLinkTimeoutGet());
629 0 : const auto start = std::chrono::steady_clock::now();
630 0 : u32 totalSentSize = 0;
631 0 : unsigned long long sentSize = 0;
632 :
633 0 : HCCL_INFO("before ra socket send, para: fdHandle[%p], data[%p], size[%u]", fdHandle, sendData, sendSize);
634 :
635 : while (true) {
636 : // 底层ra_socket_send host网卡无限制，device网卡由于HDC通道限制的限制有大小限制(目前大小为64KB)
637 0 : ret = RaSocketSend(
638 0 : fdHandle, reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(sendData) + totalSentSize),
639 0 : sendSize - totalSentSize, &sentSize);
640 0 : HCCL_INFO("ra socket send, data[%p], size[%u] send size[%u]", sendData, sendSize, totalSentSize);
641 0 : if (ret == 0) {
642 0 : totalSentSize += sentSize;
643 0 : if (totalSentSize == sendSize) { // 只有完全发送完才返回成功
644 0 : break;
645 : }
646 :
647 0 : if (totalSentSize > sendSize) {
648 0 : MACRO_THROW(
649 : NetworkApiException,
650 : StringFormat(
651 : "[Send][RaSocket]errNo[0x%016llx] ra socket send failed, fdHandle=%p, data=%p, size=%u, "
652 : "retSize=%u",
653 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), fdHandle, data, sendSize, sentSize));
654 : }
655 0 : SaluSleep(ONE_HUNDRED_MICROSECOND_OF_USLEEP);
656 0 : } else if (ret == SOCK_EAGAIN) {
657 : /* ra速率限制 retry */
658 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
659 : } else {
660 0 : MACRO_THROW(
661 : NetworkApiException,
662 : StringFormat(
663 : "[Send][RaSocket]errNo[0x%016llx] ra socket send failed, fdHandle=%p, data=%p, size=%u, "
664 : "retSize=%u, ret=%d",
665 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), fdHandle, data, sendSize, sentSize, ret));
666 : }
667 : /* 获取当前时间，如果耗时超过timeout，则返回错误 */
668 0 : const auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start);
669 0 : if (elapsed > timeout) {
670 0 : MACRO_THROW(
671 : NetworkApiException,
672 : StringFormat(
673 : "[Send][RaSocket]errNo[0x%016llx] Wait timeout for sockets send, fdHandle[%p], data[%p], size[%u], "
674 : "retsize[%u], ret[%d]",
675 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), fdHandle, data, sendSize, sentSize, ret));
676 : }
677 0 : }
678 0 : HCCL_INFO("ra socket send finished,ret[%d]", ret);
679 0 : }
680 :
681 1 : s32 HrtRaSocketNonBlockSendNormal(const FdHandle fdHandle, void* data, u64 size, u64* sentSize)
682 : {
683 2 : CHECK_NULLPTR(fdHandle, "[HrtRaSocketNonBlockSend] fdHandle is nullptr!");
684 2 : CHECK_NULLPTR(data, "[HrtRaSocketNonBlockSend] data is nullptr!");
685 1 : CHECK_NULLPTR(sentSize, "[HrtRaSocketNonBlockSend] sentSize is nullptr!");
686 3 : HCCL_INFO(
687 : "[HrtRaSocketNonBlockSend] Input params: fdHandle=%p,data=%p, size=%llu, sentSize=%llu", fdHandle, data, size,
688 : *sentSize);
689 1 : if (size > SOCKET_SEND_MAX_SIZE) {
690 0 : MACRO_THROW(
691 : NetworkApiException, StringFormat(
692 : "[hrtRaSocketNonBlockSend]errNo[0x%016llx] ra socket send size is too large, "
693 : "data[%p], size[%llu], fdHandle[%p], send size[%llu]",
694 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), data, size, fdHandle, *sentSize));
695 : }
696 :
697 1 : return RaSocketSend(fdHandle, data, size, sentSize);
698 : }
699 :
700 0 : bool HrtRaSocketNonBlockSend(const FdHandle fdHandle, void* data, u64 size, u64* sentSize)
701 : {
702 0 : s32 ret = HrtRaSocketNonBlockSendNormal(fdHandle, data, size, sentSize);
703 0 : if (ret == 0 || ret == SOCK_EAGAIN) {
704 0 : HCCL_INFO(
705 : "[HrtRaSocketNonBlockSend] ra socket send, data[%p], size[%llu], send size[%llu], ret[%d]", data, size,
706 : *sentSize, ret);
707 0 : return true;
708 : } else {
709 0 : HCCL_ERROR(
710 : "call RaSocketSend failed, fdHandle=%p, data=%p, size=%llu, sentSize=%llu, ret[%d]", fdHandle, data, size,
711 : *sentSize, ret);
712 0 : return false;
713 : }
714 : }
715 :
716 1 : HcclResult HrtRaSocketNonBlockSendHeart(const FdHandle fdHandle, void* data, u64 size, u64* sentSize)
717 : {
718 1 : s32 ret = HrtRaSocketNonBlockSendNormal(fdHandle, data, size, sentSize);
719 1 : if (ret == 0) {
720 1 : return HCCL_SUCCESS;
721 0 : } else if (ret == SOCK_EAGAIN) {
722 0 : return HCCL_E_AGAIN;
723 0 : } else if (ret == SOCK_CLOSE) {
724 0 : return HCCL_E_INTERNAL; // 暂时用这个错误码表示hccp进程异常退出
725 : } else {
726 0 : HCCL_WARNING(
727 : "[HrtRaSocketNonBlockSend]ra socket send failed, data[%p], size[%llu], send size[%llu], ret[%d]", data,
728 : size, *sentSize, ret);
729 0 : return HCCL_E_NETWORK;
730 : }
731 : }
732 :
733 1 : HcclResult HrtRaSocketNonBlockRecvHeart(const FdHandle fdHandle, void* data, u64 size, u64* recvSize)
734 : {
735 2 : CHECK_NULLPTR(fdHandle, "[HrtRaSocketNonBlockRecv] fdHandle is nullptr!");
736 2 : CHECK_NULLPTR(data, "[HrtRaSocketNonBlockRecv] data is nullptr!");
737 1 : CHECK_NULLPTR(recvSize, "[HrtRaSocketNonBlockRecv] recvSize is nullptr!");
738 3 : HCCL_DEBUG(
739 : "[HrtRaSocketNonBlockRecv] Input params: fdHandle=%p,data=%p, size=%llu, recvSize=%llu", fdHandle, data, size,
740 : *recvSize);
741 :
742 1 : s32 ret = RaSocketRecv(fdHandle, data, size, recvSize);
743 1 : if (ret == 0) {
744 1 : return HCCL_SUCCESS;
745 0 : } else if (ret == SOCK_EAGAIN) {
746 0 : return HCCL_E_AGAIN;
747 0 : } else if (ret == SOCK_CLOSE) {
748 0 : return HCCL_E_INTERNAL; // 暂时用这个错误码表示hccp进程异常退出
749 : } else {
750 0 : HCCL_WARNING(
751 : "[HrtRaSocketNonBlockRecv]ra socket recv failed, data[%p], size[%llu], "
752 : "recv[%llu], ret[%d], errno[%d][%s]",
753 : data, size, recvSize, ret, errno, strerror(errno));
754 0 : return HCCL_E_TCP_TRANSFER;
755 : }
756 : return HCCL_SUCCESS;
757 : }
758 :
759 6 : void HrtRaSocketBlockRecv(const FdHandle fdHandle, void* data, u32 size)
760 : {
761 6 : auto startTime = std::chrono::steady_clock::now();
762 6 : unsigned long long recvSize = 0;
763 6 : s32 rtRet = 0;
764 6 : u32 getedLen = 0;
765 6 : const std::chrono::seconds timeout = std::chrono::seconds(EnvLinkTimeoutGet());
766 :
767 12 : CHECK_NULLPTR(fdHandle, "[HrtRaSocketBlockRecv] fdHandle is nullptr!");
768 6 : CHECK_NULLPTR(data, "[HrtRaSocketBlockRecv] data is nullptr!");
769 18 : HCCL_INFO("before ra socket recv, para: fdHandle[%p], data[%p], size[%u]", fdHandle, data, size);
770 : while (true) {
771 6 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
772 : std::string errMsg = StringFormat(
773 : "[Recv][RaSocket]errNo[0x%016llx] Wait timeout for sockets recv, data[%p], "
774 : "size[%u], recvSize[%u], fdHandle[%p], ret[%d]",
775 1 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), data, size, recvSize, fdHandle, rtRet);
776 3 : HCCL_ERROR("%s", errMsg.c_str());
777 3 : HCCL_ERROR("Please check the following reasons:");
778 3 : HCCL_ERROR("1. check the firewall configuration or try to disable the firewall.");
779 3 : HCCL_ERROR("2. check error log on the other process or thread.");
780 4 : MACRO_THROW(NetworkApiException, errMsg);
781 1 : }
782 10 : rtRet = RaSocketRecv(
783 5 : fdHandle, reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(data) + getedLen), size - getedLen,
784 : &recvSize);
785 5 : if ((rtRet == 0) && (recvSize > 0)) { // 接收完成，也有可能要多次接收
786 2 : getedLen += recvSize;
787 2 : if (getedLen > size) {
788 4 : MACRO_THROW(
789 : NetworkApiException,
790 : StringFormat(
791 : "[Recv][RaSocket]errNo[0x%016llx] socket receive call RaSocketRecv failed,"
792 : "rtSize[%u], bigger size[%zu], fdHandle[%p], data[%p], retSize[%u], ret[%d]",
793 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_TRANSFER), getedLen, size, fdHandle, data, recvSize,
794 : rtRet));
795 : }
796 1 : if (getedLen == size) {
797 1 : break;
798 : }
799 3 : } else if ((rtRet == 0) && (recvSize == 0)) {
800 4 : MACRO_THROW(
801 : NetworkApiException,
802 : StringFormat(
803 : "[Recv][RaSocket]recv fail, fdHandle=%p, data=%p, bufLen=%u, recLen=%lld, ret=%d", fdHandle, data,
804 : size, recvSize, rtRet));
805 2 : } else if (rtRet == SOCK_ESOCKCLOSED || rtRet == SOCK_CLOSE) { // 连接关闭，出错
806 8 : MACRO_THROW(
807 : NetworkApiException,
808 : StringFormat(
809 : "[Recv][RaSocket]errNo[0x%016llx] recv fail, call RaSocketRecv failed, sock_esockclosed, "
810 : "fdhandle=%p, data=%p, bufLen=%u, recLen=%lld, ret=%d",
811 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_TRANSFER), fdHandle, data, size, recvSize, rtRet));
812 0 : } else if (rtRet != 0) {
813 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP); // 尚未接收到数据,延时1ms
814 0 : continue;
815 : }
816 0 : }
817 3 : HCCL_INFO("ra socket receive finished. ret[%d]", rtRet);
818 1 : }
819 :
820 2 : SocketHandle HrtRaSocketInit(HrtNetworkMode netMode, RaInterface& in)
821 : {
822 2 : int mode = HRT_NETWORK_MODE_MAP.at(netMode);
823 2 : struct rdev rdevInfo {};
824 2 : rdevInfo.phyId = in.phyId;
825 2 : rdevInfo.family = in.address.GetFamily();
826 2 : rdevInfo.localIp = IpAddressToHccpIpAddr(in.address);
827 :
828 6 : HCCL_INFO(
829 : "[HrtRaSocketInit] Input params: mode=%u, ip=%u, device id=%u, family=%u", mode, rdevInfo.localIp.addr.s_addr,
830 : rdevInfo.phyId, rdevInfo.family);
831 :
832 2 : SocketHandle socketHandle = nullptr;
833 2 : s32 ret = RaSocketInit(mode, rdevInfo, &socketHandle);
834 2 : if (ret != 0 || (socketHandle == nullptr)) {
835 4 : MACRO_THROW(
836 : NetworkApiException, StringFormat(
837 : "[Init][RaSock]errNo[0x%016llx] ra socket init fail, call RaSocketInit failed, "
838 : "params: mode=%u, ip=%u, device id=%u, family=%u. return: ret=%d",
839 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), mode, rdevInfo.localIp.addr.s_addr,
840 : rdevInfo.phyId, rdevInfo.family, ret));
841 : }
842 :
843 3 : HCCL_INFO("socket init success, ip[%u], device id[%u], ret[%d]", rdevInfo.localIp.addr.s_addr, rdevInfo.phyId, ret);
844 1 : return socketHandle;
845 : }
846 :
847 2 : void HrtRaSocketDeInit(SocketHandle socketHandle)
848 : {
849 2 : CHECK_NULLPTR(socketHandle, "[HrtRaSocketDeInit] socketHandle is nullptr!");
850 6 : HCCL_INFO("[HrtRaSocketDeInit] Input params: socketHandle=%p", socketHandle);
851 :
852 2 : s32 ret = RaSocketDeinit(socketHandle);
853 2 : if (ret != 0) {
854 0 : MACRO_THROW(
855 : NetworkApiException, StringFormat(
856 : "[DeInit][RaSocket]errNo[0x%016llx] rt socket deinit fail. call RaSocketDeinit "
857 : "failed, params: socketHandle[%p], return: ret[%d]",
858 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), socketHandle, ret));
859 : }
860 2 : }
861 :
862 0 : void HrtRaSocketSetWhiteListStatus(u32 enable)
863 : {
864 0 : HCCL_INFO("[HrtRaSocketSetWhiteListStatus] Input params: enable=%u", enable);
865 0 : s32 ret = RaSocketSetWhiteListStatus(enable);
866 0 : if (ret != 0) {
867 0 : MACRO_THROW(
868 : NetworkApiException, StringFormat(
869 : "[Set][WhiteListStatus]errNo[0x%016llx] ra socekt set white list fail, call "
870 : "RaSocketSetWhiteListStatus failed, params: enable[%u], return: ret[%d]",
871 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), enable, ret));
872 : }
873 :
874 0 : HCCL_INFO("set host socket whitelist status[%u] success.", enable);
875 0 : }
876 :
877 0 : u32 HrtRaSocketGetWhiteListStatus()
878 : {
879 : u32 enable;
880 0 : s32 ret = RaSocketGetWhiteListStatus(&enable);
881 0 : if (ret != 0) {
882 0 : MACRO_THROW(
883 : NetworkApiException, StringFormat(
884 : "[Get][WhiteListStatus]errNo[0x%016llx] ra socekt get whilte list fail, call "
885 : "RaSocketGetWhiteListStatus failed, return: ret[%d]",
886 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret));
887 : }
888 :
889 0 : HCCL_INFO("get host socket whitelist status[%u] success.", enable);
890 0 : return enable;
891 : }
892 :
893 5 : void HrtRaSocketWhiteListAdd(SocketHandle socketHandle, vector<RaSocketWhitelist>& wlists)
894 : {
895 5 : CHECK_NULLPTR(socketHandle, "[HrtRaSocketWhiteListAdd] socketHandle is nullptr!");
896 15 : HCCL_INFO("[HrtRaSocketWhiteListAdd] Input params: socketHandle=%p", socketHandle);
897 :
898 5 : vector<struct SocketWlistInfoT> wlistInfoVec;
899 5 : wlistInfoVec.reserve(MAX_NUM_OF_WHITE_LIST_NUM);
900 5 : size_t wlistNum = wlists.size();
901 5 : size_t startIdx = 0;
902 8 : while (wlistNum > 0) {
903 5 : size_t addListNum = wlistNum > MAX_NUM_OF_WHITE_LIST_NUM ? MAX_NUM_OF_WHITE_LIST_NUM : wlistNum;
904 9 : for (size_t idx = startIdx; idx < addListNum + startIdx; idx++) {
905 5 : struct SocketWlistInfoT wlistInfo {};
906 5 : wlistInfo.connLimit = wlists[idx].connLimit;
907 5 : wlistInfo.remoteIp = IpAddressToHccpIpAddr(wlists[idx].remoteIp);
908 :
909 5 : int sret = strcpy_s(wlistInfo.tag, sizeof(wlistInfo.tag), wlists[idx].tag.c_str());
910 5 : if (sret != EOK) {
911 4 : MACRO_THROW(
912 : InternalException,
913 : StringFormat(
914 : "[Add][RaSocketWhiteList]errNo[0x%016llx]errName[HCCL_E_MEMORY] memory copy failed. params: "
915 : "socketHandle[%p], return: ret[%d], wlistInfo.tag size=%zu, wlists[%zu].tag size=%zu",
916 : HCOM_ERROR_CODE(HcclResult::HCCL_E_MEMORY), socketHandle, sret, sizeof(wlistInfo.tag), idx,
917 : sizeof(wlists[idx].tag.c_str())));
918 : }
919 12 : HCCL_INFO(
920 : "add whitelistInfo tag=[%s], remoteIp[%s]", wlistInfo.tag, wlists[idx].remoteIp.Describe().c_str());
921 4 : wlistInfoVec.push_back(wlistInfo);
922 : }
923 :
924 4 : s32 ret = RaSocketWhiteListAdd(socketHandle, wlistInfoVec.data(), wlistInfoVec.size());
925 4 : if (ret != 0) {
926 4 : MACRO_THROW(
927 : NetworkApiException,
928 : StringFormat(
929 : "[Add][RaSocketWhiteList]errNo[0x%016llx]errName[HCCL_E_TCP_CONNECT] ra white list add fail, call "
930 : "RaSocketWhiteListAdd failed, socketHandle[%p], num=%zu, return[%d].",
931 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), socketHandle, wlistInfoVec.size() + startIdx,
932 : ret));
933 : }
934 9 : HCCL_INFO("add white list: num[%zu], remain [%zu].", addListNum, (wlistNum - addListNum));
935 :
936 3 : wlistInfoVec.clear();
937 3 : wlistNum -= addListNum;
938 3 : startIdx += addListNum;
939 : }
940 9 : HCCL_INFO("[HrtRaSocketWhiteListAdd] Success. Total add num [%zu]", wlists.size());
941 5 : }
942 :
943 4 : void HrtRaSocketWhiteListDel(SocketHandle socketHandle, vector<RaSocketWhitelist>& wlists)
944 : {
945 4 : CHECK_NULLPTR(socketHandle, "[HrtRaSocketWhiteListDel] socketHandle is nullptr!");
946 12 : HCCL_INFO("[HrtRaSocketWhiteListDel] Input params: socketHandle=%p", socketHandle);
947 :
948 4 : vector<struct SocketWlistInfoT> wlistInfoVec;
949 4 : wlistInfoVec.reserve(MAX_NUM_OF_WHITE_LIST_NUM);
950 4 : size_t wlistNum = wlists.size();
951 4 : size_t startIdx = 0;
952 10 : while (wlistNum > 0) {
953 7 : size_t delListNum = wlistNum > MAX_NUM_OF_WHITE_LIST_NUM ? MAX_NUM_OF_WHITE_LIST_NUM : wlistNum;
954 66 : for (size_t idx = startIdx; idx < delListNum + startIdx; idx++) {
955 59 : struct SocketWlistInfoT wlistInfo {};
956 59 : wlistInfo.connLimit = wlists[idx].connLimit;
957 59 : wlistInfo.remoteIp = IpAddressToHccpIpAddr(wlists[idx].remoteIp);
958 :
959 59 : int sret = strcpy_s(wlistInfo.tag, sizeof(wlistInfo.tag), wlists[idx].tag.c_str());
960 59 : if (sret != EOK) {
961 : auto msg = StringFormat(
962 : "[Del][RaSocketWhiteList]errNo[0x%016llx] memory copy failed. ret[%d], wlistInfo.tag size[%zu], "
963 : "wlists[%zu].tag size[%zu]",
964 : HCOM_ERROR_CODE(HcclResult::HCCL_E_MEMORY), sret, sizeof(wlistInfo.tag), idx,
965 0 : sizeof(wlists[idx].tag.c_str()));
966 0 : MACRO_THROW(InternalException, msg);
967 0 : }
968 59 : wlistInfoVec.push_back(wlistInfo);
969 : }
970 :
971 7 : s32 ret = RaSocketWhiteListDel(socketHandle, wlistInfoVec.data(), wlistInfoVec.size());
972 7 : if (ret != 0) {
973 4 : MACRO_THROW(
974 : NetworkApiException, StringFormat(
975 : "[Del][RaSocketWhiteList]errNo[0x%016llx] ra white list del fail, call "
976 : "RaSocketWhiteListDel failed, num=%zu, return[%d].",
977 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), wlists.size(), ret));
978 : }
979 18 : HCCL_INFO("del white list: num[%zu], remain [%zu].", delListNum, (wlistNum - delListNum));
980 :
981 6 : wlistInfoVec.clear();
982 6 : wlistNum -= delListNum;
983 6 : startIdx += delListNum;
984 : }
985 9 : HCCL_INFO("[HrtRaSocketWhiteListDel] Success. Total delete num[%zu]", wlists.size());
986 4 : }
987 :
988 : std::mutex g_deviceVnicIpMutex;
989 : std::map<u32, IpAddress> g_deviceIdVnicInfoMap; // 记录deviceid和vnic ip的关系，用于server内查询，避免重复查询
990 :
991 0 : void HrtRaSocketGetVnicIpInfos(u32 phyId, DeviceIdType deviceIdType, u32 deviceId, IpAddress& vnicIP)
992 : {
993 0 : std::lock_guard<std::mutex> lock(g_deviceVnicIpMutex);
994 0 : auto iter = g_deviceIdVnicInfoMap.find(deviceId);
995 0 : if (iter != g_deviceIdVnicInfoMap.end()) {
996 : // 缓存查找到，直接从缓存获取
997 0 : vnicIP = iter->second;
998 0 : HCCL_INFO(
999 : "[HrtRaSocketGetVnicIpInfos] vnicInfoMap deviceId[%u] found, Ip[%s]", deviceId, vnicIP.Describe().c_str());
1000 0 : return;
1001 : }
1002 0 : struct IpInfo vnicIpInfo = {};
1003 0 : (void)memset_s(&vnicIpInfo, sizeof(IpInfo), 0, sizeof(IpInfo));
1004 0 : IdType idType = static_cast<IdType>(deviceIdType);
1005 0 : auto ret = RaSocketGetVnicIpInfos(phyId, idType, &deviceId, 1, &vnicIpInfo);
1006 0 : if (ret != 0) {
1007 0 : HCCL_ERROR("[hrtRaGetSocketVnicIpInfo]ra get VnicIpfail. ret[%d]", ret);
1008 0 : throw NetworkApiException(StringFormat("call hrtRaGetSocketVnicIpInfo failed, ret=%llu", ret));
1009 : }
1010 : BinaryAddr temp;
1011 0 : temp.addr = vnicIpInfo.ip.addr;
1012 0 : temp.addr6 = vnicIpInfo.ip.addr6;
1013 0 : IpAddress ipInfo(temp, vnicIpInfo.family);
1014 0 : if (ipInfo.IsInvalid()) {
1015 0 : HCCL_ERROR("vnicIp is invalid.");
1016 0 : throw NetworkApiException("vnicIp is invalid.");
1017 : }
1018 0 : g_deviceIdVnicInfoMap.insert({deviceId, ipInfo});
1019 0 : vnicIP = ipInfo;
1020 0 : HCCL_INFO(
1021 : "[hrtRaGetSocketVnicIpInfos] add vnicInfoMap, deviceIds[%u], Ip[%s]", deviceId, vnicIP.Describe().c_str());
1022 0 : }
1023 :
1024 7 : static u32 HrtGetIfNum(struct RaGetIfattr& config)
1025 : {
1026 21 : HCCL_INFO("[HrtGetIfNum] Input params: phyId=%u, nicPosistion=%u", config.phyId, config.nicPosition);
1027 :
1028 7 : u32 num = 0;
1029 7 : s32 ret = RaGetIfnum(&config, &num);
1030 7 : if (ret != 0) {
1031 4 : MACRO_THROW(
1032 : NetworkApiException,
1033 : StringFormat(
1034 : "[Get][IfNum]errNo[0x%016llx] ra get if num fail. call RaGetIfnum failed, Input params: phyId=%u, "
1035 : "nicPosistion=%u, return: ret[%d], num[%u]",
1036 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), config.phyId, config.nicPosition, ret, num));
1037 : }
1038 6 : return num;
1039 : }
1040 :
1041 3 : static void HrtGetIfAddress(struct RaGetIfattr& config, InterfaceInfo ifaddrInfos[], u32& num)
1042 : {
1043 3 : CHECK_NULLPTR(ifaddrInfos, "[HrtGetIfAddress] ifaddrInfos is nullptr!");
1044 9 : HCCL_INFO(
1045 : "[HrtGetIfAddress] Input params: phyId=%u, nicPosition=%u, num=%u", config.phyId, config.nicPosition, num);
1046 :
1047 3 : s32 ret = RaGetIfaddrs(&config, ifaddrInfos, &num);
1048 3 : if (ret != 0) {
1049 4 : MACRO_THROW(
1050 : NetworkApiException,
1051 : StringFormat(
1052 : "[Get][IfAddress]errNo[0x%016llx] ra get if address fail. call RaGetIfaddrs failed, Input params: "
1053 : "phyId=%u, nicPosistion=%u, return: ret[%d], num[%u]",
1054 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), config.phyId, config.nicPosition, ret, num));
1055 : }
1056 2 : }
1057 :
1058 4 : std::vector<std::pair<std::string, IpAddress>> HrtGetHostIf(u32 devPhyId)
1059 : {
1060 12 : HCCL_INFO("[HrtGetHostIf] Input params: devPhyId=%u", devPhyId);
1061 4 : std::vector<std::pair<std::string, IpAddress>> hostIfs;
1062 4 : struct RaGetIfattr config = {};
1063 4 : config.phyId = devPhyId;
1064 4 : config.nicPosition = static_cast<u32>(NetworkMode::NETWORK_PEER_ONLINE);
1065 :
1066 4 : u32 ifAddrNum = HrtGetIfNum(config);
1067 9 : HCCL_RUN_INFO("[Get][HostIf]hrtGetIfNum success. ifAddrNum[%u].", ifAddrNum);
1068 3 : if (ifAddrNum == 0) {
1069 3 : HCCL_WARNING("[Get][HostIf]there is no valid host interface, ifAddrNum[%u].", ifAddrNum);
1070 1 : return hostIfs;
1071 : }
1072 :
1073 : std::shared_ptr<struct InterfaceInfo> ifAddrInfoPtrs(
1074 4 : new InterfaceInfo[ifAddrNum](), std::default_delete<InterfaceInfo[]>());
1075 2 : struct InterfaceInfo* ifAddrInfos = ifAddrInfoPtrs.get();
1076 :
1077 2 : (void)memset_s(ifAddrInfos, ifAddrNum * sizeof(InterfaceInfo), 0, ifAddrNum * sizeof(InterfaceInfo));
1078 :
1079 2 : HrtGetIfAddress(config, ifAddrInfos, ifAddrNum);
1080 :
1081 2 : for (u32 i = 0; i < ifAddrNum; i++) {
1082 1 : IpAddress ip = IfAddrInfoToIpAddress(ifAddrInfos[i]);
1083 1 : hostIfs.emplace_back(ifAddrInfos[i].ifname, ip);
1084 3 : HCCL_INFO("HrtGetIfAddress: idx[%u], ifName[%s], ip[%s]", i, ifAddrInfos[i].ifname, ip.GetIpStr().c_str());
1085 : }
1086 :
1087 1 : return hostIfs;
1088 4 : }
1089 :
1090 3 : vector<IpAddress> HrtGetDeviceIp(u32 devicePhyId, NetworkMode netWorkMode)
1091 : {
1092 9 : HCCL_INFO("[HrtGetDeviceIp] Input params: devicePhyId=%u", devicePhyId);
1093 3 : vector<IpAddress> ipAddr;
1094 3 : struct RaGetIfattr config = {};
1095 3 : config.phyId = devicePhyId;
1096 3 : config.nicPosition = static_cast<u32>(netWorkMode);
1097 :
1098 3 : u32 ifAddrNum = HrtGetIfNum(config);
1099 9 : HCCL_RUN_INFO("[Get][DeviceIP]hrtGetIfNum success. ifAddrNum[%u].", ifAddrNum);
1100 :
1101 3 : if (ifAddrNum == 0) {
1102 6 : HCCL_WARNING("[Get][DeviceIP]device has no ip information, phy_id[%u]", devicePhyId);
1103 2 : return ipAddr;
1104 : }
1105 :
1106 : std::shared_ptr<struct InterfaceInfo> ifAddrInfoPtrs(
1107 2 : new InterfaceInfo[ifAddrNum](), std::default_delete<InterfaceInfo[]>());
1108 1 : struct InterfaceInfo* ifAddrInfos = ifAddrInfoPtrs.get();
1109 :
1110 1 : (void)memset_s(ifAddrInfos, ifAddrNum * sizeof(InterfaceInfo), 0, ifAddrNum * sizeof(InterfaceInfo));
1111 :
1112 1 : HrtGetIfAddress(config, ifAddrInfos, ifAddrNum);
1113 :
1114 2 : for (u32 i = 0; i < ifAddrNum; i++) {
1115 1 : IpAddress ip = IfAddrInfoToIpAddress(ifAddrInfos[i]);
1116 1 : ipAddr.emplace_back(ip);
1117 3 : HCCL_INFO("HrtGetIfAddress: idx[%u], ifName[%s], ip[%s]", i, ifAddrInfos[i].ifname, ip.GetIpStr().c_str());
1118 : }
1119 :
1120 1 : return ipAddr;
1121 1 : }
1122 :
1123 2 : RdmaHandle HrtRaRdmaInit(HrtNetworkMode netMode, RaInterface& in)
1124 : {
1125 2 : RdmaHandle rdmaHandle = nullptr;
1126 2 : int mode = HRT_NETWORK_MODE_MAP.at(netMode);
1127 2 : unsigned int notifyType = netMode == HrtNetworkMode::PEER ? NO_USE : NOTIFY;
1128 6 : HCCL_INFO("[HrtRaRdmaInit] Input params: mode=%d, phyId=%u", mode, in.phyId);
1129 2 : struct rdev rdevInfo {};
1130 2 : rdevInfo.phyId = in.phyId;
1131 2 : rdevInfo.family = in.address.GetFamily();
1132 2 : rdevInfo.localIp = IpAddressToHccpIpAddr(in.address);
1133 2 : s32 ret = RaRdevInit(mode, notifyType, rdevInfo, &rdmaHandle);
1134 12 : RPT_INPUT_ERR(
1135 : ret == HCCP_ELINKDOWN, "EI0009", vector<string>({"device_id", "reason"}),
1136 : vector<string>({std::to_string(rdevInfo.phyId), "The network port is down"}));
1137 2 : if (ret != 0 || (rdmaHandle == nullptr)) {
1138 8 : MACRO_THROW(
1139 : NetworkApiException, StringFormat(
1140 : "[Init][RaRdma]errNo[0x%016llx] rdma init fail. call RaRdevInit failed, Input "
1141 : "params: phyId=%u, mode=%u, return: ret[%d]",
1142 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), in.phyId, mode, ret));
1143 : }
1144 0 : return rdmaHandle;
1145 2 : }
1146 :
1147 12 : void HrtRaRdmaDeInit(RdmaHandle rdmaHandle, HrtNetworkMode netMode)
1148 : {
1149 12 : CHECK_NULLPTR(rdmaHandle, "[HrtRaRdmaDeInit] rdmaHandle is nullptr!");
1150 36 : HCCL_INFO("[HrtRaRdmaDeInit] Input params: rdmaHandle=%p, netMode=%d", rdmaHandle, netMode);
1151 12 : unsigned int notifyType = netMode == HrtNetworkMode::PEER ? NO_USE : NOTIFY;
1152 12 : s32 ret = RaRdevDeinit(rdmaHandle, notifyType);
1153 12 : if (ret != 0) {
1154 0 : MACRO_THROW(
1155 : NetworkApiException, StringFormat(
1156 : "[DeInit][RaRdma]errNo[0x%016llx] rt rdev deinit fail. call RaRdevDeinit failed, "
1157 : "rdmaHandle=%p, return[%d].",
1158 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), rdmaHandle, ret));
1159 : }
1160 12 : }
1161 :
1162 0 : void HrtRaGetNotifyBaseAddr(RdmaHandle rdmaHandle, u64* va, u64* size)
1163 : {
1164 0 : CHECK_NULLPTR(rdmaHandle, "[HrtRaGetNotifyBaseAddr] rdmaHandle is nullptr!");
1165 0 : CHECK_NULLPTR(va, "[HrtRaGetNotifyBaseAddr] va is nullptr!");
1166 0 : CHECK_NULLPTR(size, "[HrtRaGetNotifyBaseAddr] size is nullptr!");
1167 :
1168 0 : HCCL_INFO("[HrtRaGetNotifyBaseAddr] Input params: rdmaHandle=%p, va=%llu, size=%llu", rdmaHandle, *va, *size);
1169 0 : auto startTime = std::chrono::steady_clock::now();
1170 0 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
1171 : while (true) {
1172 : unsigned long long notifyVa;
1173 : unsigned long long notifySize;
1174 0 : s32 ret = RaGetNotifyBaseAddr(rdmaHandle, ¬ifyVa, ¬ifySize);
1175 0 : if (ret == 0) {
1176 0 : *va = notifyVa;
1177 0 : *size = notifySize;
1178 0 : break;
1179 0 : } else if (ret == SOCK_EAGAIN) {
1180 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
1181 0 : if (bTimeout != 0) {
1182 0 : HCCL_ERROR(
1183 : "[Get][RaNotifyBaseAddr]errNo[0x%016llx] ra get notify base addr "
1184 : "timeout[%lld s]. return[%d], params: rdmaHandle[%p], va[0x%llx], size[%llu]",
1185 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), timeout, ret, rdmaHandle, notifyVa, notifySize);
1186 : }
1187 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
1188 : } else {
1189 0 : MACRO_THROW(
1190 : NetworkApiException,
1191 : StringFormat(
1192 : "[Get][RaNotifyBaseAddr]errNo[0x%016llx] ra get notify base addr fail, call RaGetNotifyBaseAddr "
1193 : "failed,"
1194 : "return[%d], params: va[0x%llx], size[%llu], rdmaHandle=%p",
1195 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), ret, notifyVa, notifySize, rdmaHandle));
1196 : }
1197 0 : }
1198 0 : }
1199 :
1200 1 : QpHandle HrtRaQpCreate(RdmaHandle rdmaHandle, int flag, int qpMode)
1201 : {
1202 1 : CHECK_NULLPTR(rdmaHandle, "[HrtRaQpCreate] rdmaHandle is nullptr!");
1203 3 : HCCL_INFO("[HrtRaQpCreate] Input params: rdmaHandle=%p, flag=%d, qpMode=%d", rdmaHandle, flag, qpMode);
1204 1 : QpHandle connHandle = nullptr;
1205 :
1206 1 : s32 ret = RaQpCreate(rdmaHandle, flag, qpMode, &connHandle);
1207 1 : if (ret != 0 || connHandle == nullptr) {
1208 1 : RPT_INPUT_ERR(
1209 : ret == ROCE_ENOMEM_RET, "EI0011",
1210 : std::vector<std::string>({"memory_size"}), // A3是当ROCE_ENOMEM_RET才上报EI0011,内存大小取决于qp深度配置
1211 : std::vector<std::string>({"262144~3145728"}));
1212 4 : MACRO_THROW(
1213 : NetworkApiException,
1214 : StringFormat(
1215 : "[Create][RaQp]errNo[0x%016llx] ra qp create fail. call RaGetNotifyBaseAddr, params: rdmaHandle[%p], "
1216 : "flag[%d], qpMode[%d], connHandle[%p]. return: ret[%d]",
1217 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), rdmaHandle, flag, qpMode, connHandle, ret));
1218 : }
1219 0 : return connHandle;
1220 : }
1221 :
1222 9 : void HrtRaQpDestroy(QpHandle qpHandle)
1223 : {
1224 9 : CHECK_NULLPTR(qpHandle, "[HrtRaQpDestroy] qpHandle is nullptr!");
1225 27 : HCCL_INFO("[HrtRaQpDestroy] Input params: qpHandle=%p", qpHandle);
1226 9 : auto startTime = std::chrono::steady_clock::now();
1227 9 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
1228 : while (true) {
1229 9 : s32 ret = RaQpDestroy(qpHandle);
1230 9 : if (ret == 0) {
1231 9 : break;
1232 0 : } else if (ret == SOCK_EAGAIN) {
1233 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
1234 0 : if (bTimeout != 0) {
1235 0 : MACRO_THROW(
1236 : NetworkApiException, StringFormat(
1237 : "[Destroy][RaQp]errNo[0x%016llx] ra qp destroy timeout[%d s]. "
1238 : "qpHandle[%p], return[%d].",
1239 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), timeout, qpHandle, ret));
1240 : }
1241 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
1242 : } else {
1243 0 : MACRO_THROW(
1244 : NetworkApiException, StringFormat(
1245 : "[Destroy][RaQp]errNo[0x%016llx] ra qp destroy fail. call RaQpDestroy failed, "
1246 : "qpHandle[%p], return[%d].",
1247 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), qpHandle, ret));
1248 : }
1249 0 : }
1250 9 : }
1251 :
1252 0 : void HrtRaQpConnectAsync(QpHandle qpHandle, FdHandle fdHandle)
1253 : {
1254 0 : CHECK_NULLPTR(qpHandle, "[HrtRaQpConnectAsync] qpHandle is nullptr!");
1255 0 : CHECK_NULLPTR(fdHandle, "[HrtRaQpConnectAsync] fdHandle is nullptr!");
1256 :
1257 0 : HCCL_INFO("[HrtRaQpConnectAsync] Input params: qpHandle=%p, fdHandle=%p", qpHandle, fdHandle);
1258 0 : auto startTime = std::chrono::steady_clock::now();
1259 0 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
1260 : while (true) {
1261 0 : s32 ret = RaQpConnectAsync(qpHandle, fdHandle);
1262 0 : if (ret == 0) {
1263 0 : break;
1264 0 : } else if (ret == SOCK_EAGAIN) {
1265 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
1266 0 : if (bTimeout != 0) {
1267 0 : HCCL_ERROR(
1268 : "[ConnectAsync][RaQp]errNo[0x%016llx] ra qp connect async "
1269 : "timeout[%lld s]. qpHandle=[%p], fdHandle=[%p], return[%d].",
1270 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), timeout, qpHandle, fdHandle, ret);
1271 : }
1272 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
1273 : } else {
1274 0 : MACRO_THROW(
1275 : NetworkApiException, StringFormat(
1276 : "[ConnectAsync][RaQp]errNo[0x%016llx] ra qp connect async fail. call "
1277 : "RaQpConnectAsync failed, qpHandle=%p, fdHandle=%p, return[%d]",
1278 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), qpHandle, fdHandle, ret));
1279 : }
1280 0 : }
1281 0 : }
1282 :
1283 1 : int HrtGetRaQpStatus(QpHandle qpHandle)
1284 : {
1285 1 : CHECK_NULLPTR(qpHandle, "[HrtGetRaQpStatus] qpHandle is nullptr!");
1286 3 : HCCL_INFO("[HrtGetRaQpStatus] Input params: qpHandle=%p", qpHandle);
1287 1 : int status = 0;
1288 1 : s32 ret = RaGetQpStatus(qpHandle, &status);
1289 1 : if (ret != 0) {
1290 4 : MACRO_THROW(
1291 : NetworkApiException, StringFormat(
1292 : "[GetStatus][RaQp]errNo[0x%016llx] ra qp get status failed. call ra_get_status "
1293 : "failed, qpHandle[%p], return[%d]",
1294 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), qpHandle, ret));
1295 : }
1296 0 : return status;
1297 : }
1298 :
1299 1 : void HrtRaMrReg(QpHandle qpHandle, RaMrInfo& info)
1300 : {
1301 1 : CHECK_NULLPTR(qpHandle, "[HrtRaMrReg] qpHandle is nullptr!");
1302 1 : struct MrInfoT mrInfo = {};
1303 1 : mrInfo.addr = info.addr;
1304 1 : mrInfo.size = info.size;
1305 1 : mrInfo.access = info.access;
1306 1 : mrInfo.lkey = info.lkey;
1307 3 : HCCL_INFO(
1308 : "ra mr reg: qpHandle[%p], addr[%p], size[%llu], access[%d]", qpHandle, mrInfo.addr, mrInfo.size, mrInfo.access);
1309 1 : s32 ret = RaMrReg(qpHandle, &mrInfo);
1310 1 : if (ret != 0) {
1311 4 : MACRO_THROW(
1312 : NetworkApiException,
1313 : StringFormat(
1314 : "[Reg][RaMr]errNo[0x%016llx] ra mr reg fail. call RaMrReg failed, return[%d], params: qpHandle[%p], "
1315 : "addr[%p], size[%llu], access[%d]",
1316 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), ret, qpHandle, mrInfo.addr, mrInfo.size, mrInfo.access));
1317 : }
1318 0 : }
1319 :
1320 1 : void HrtRaMrDereg(QpHandle qpHandle, RaMrInfo& info)
1321 : {
1322 1 : CHECK_NULLPTR(qpHandle, "[HrtRaMrDereg] qpHandle is nullptr!");
1323 1 : struct MrInfoT mrInfo = {};
1324 1 : mrInfo.addr = info.addr;
1325 1 : mrInfo.size = info.size;
1326 1 : mrInfo.access = info.access;
1327 1 : mrInfo.lkey = info.lkey;
1328 3 : HCCL_INFO(
1329 : "ra mr dereg: qpHandle[%p], addr[%p], size[%llu], access[%d]", qpHandle, mrInfo.addr, mrInfo.size,
1330 : mrInfo.access);
1331 1 : s32 ret = RaMrDereg(qpHandle, &mrInfo);
1332 1 : if (ret != 0) {
1333 : string msg = StringFormat(
1334 : "call RaMrDereg failed, qpHandle=%p, addr=%p, size=%llu, access=%d", qpHandle, mrInfo.addr, mrInfo.size,
1335 1 : mrInfo.access);
1336 4 : MACRO_THROW(NetworkApiException, msg);
1337 1 : }
1338 0 : }
1339 :
1340 3 : static void HrtRaSendWr(QpHandle qpHandle, struct SendWr* wr, struct SendWrRsp* opRsp)
1341 : {
1342 6 : CHECK_NULLPTR(qpHandle, "[HrtRaSendWr] qpHandle is nullptr!");
1343 6 : CHECK_NULLPTR(wr, "[HrtRaSendWr] wr is nullptr!");
1344 3 : CHECK_NULLPTR(opRsp, "[HrtRaSendWr] opRsp is nullptr!");
1345 9 : HCCL_INFO("[HrtRaSendWr] Input params: qpHandle=%p, send_wrAddr=%p, opRspAddr=%p", qpHandle, wr, opRsp);
1346 3 : auto startTime = std::chrono::steady_clock::now();
1347 3 : auto timeout = std::chrono::seconds(EnvLinkTimeoutGet());
1348 : while (true) {
1349 3 : s32 ret = RaSendWr(qpHandle, wr, opRsp);
1350 3 : if (ret == 0) {
1351 3 : break;
1352 0 : } else if (ret == SOCK_ENOENT || ret == SOCK_EAGAIN) {
1353 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout);
1354 0 : if (bTimeout) {
1355 0 : HCCL_ERROR(
1356 : "[Send][RaWr]errNo[0x%016llx] ra get send async timeout[%d s]. "
1357 : "return[%d], params: qpHandle[%p], send_wrAddr[%p], opRspAddr[%p]",
1358 : HCCL_ERROR_CODE(HcclResult::HCCL_E_ROCE_TRANSFER), timeout, ret, qpHandle, wr, opRsp);
1359 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
1360 : }
1361 0 : } else {
1362 : string msg
1363 0 : = StringFormat("call RaSendWr failed, qpHandle=%p, send_wrAddr=%p opRspAddr=%p", qpHandle, wr, opRsp);
1364 0 : MACRO_THROW(NetworkApiException, msg);
1365 0 : }
1366 0 : }
1367 3 : }
1368 :
1369 3 : RaSendWrResp HrtRaSendOneWr(QpHandle qpHandle, HRaSendWr& in)
1370 : {
1371 3 : CHECK_NULLPTR(qpHandle, "[HrtRaSendOneWr] qpHandle is nullptr!");
1372 9 : HCCL_INFO(
1373 : "[HrtRaSendOneWr] Input params: qpHandle=%p, locAddr=0x%llx, len=%u, rmtAddr=0x%llx, op=%u, sendFlag=%d",
1374 : qpHandle, in.locAddr, in.len, in.rmtAddr, in.op, in.sendFlag);
1375 3 : struct SgList bufList {};
1376 3 : bufList.addr = in.locAddr;
1377 3 : bufList.len = in.len;
1378 :
1379 3 : struct SendWr wr = {};
1380 3 : wr.op = in.op;
1381 3 : wr.dstAddr = in.rmtAddr;
1382 3 : wr.sendFlag = in.sendFlag;
1383 3 : wr.bufNum = 1; // 此处list只有一个，设置为1
1384 3 : wr.bufList = &bufList;
1385 3 : struct SendWrRsp opRsp = {};
1386 3 : HrtRaSendWr(qpHandle, &wr, &opRsp);
1387 :
1388 6 : return RaSendWrResp(opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex, opRsp.db.dbIndex, opRsp.db.dbInfo);
1389 : }
1390 :
1391 0 : string HrtRaGetKeyDescribe(const u8* key, u32 len)
1392 : {
1393 0 : CHECK_NULLPTR(key, "[HrtRaGetKeyDescribe] key is nullptr!");
1394 0 : HCCL_INFO("[HrtRaGetKeyDescribe] Input params: key=%d, len=%u", *key, len);
1395 0 : string desc = "0x";
1396 0 : for (u32 idx = 0; idx < len; idx++) {
1397 0 : desc += StringFormat("%02x", key[idx]);
1398 : }
1399 0 : return desc;
1400 0 : }
1401 :
1402 11 : RdmaHandle HrtRaUbCtxInit(const HrtRaUbCtxInitParam& in)
1403 : {
1404 33 : HCCL_INFO(
1405 : "[HrtRaUbCtxInit] Input params: mode=%d, phyId=%u, addr=%s", in.mode, in.phyId, in.addr.GetIpStr().c_str());
1406 11 : struct CtxInitCfg initCfg {};
1407 11 : initCfg.mode = HRT_NETWORK_MODE_MAP.at(in.mode);
1408 :
1409 11 : struct CtxInitAttr ctxInfo {};
1410 11 : ctxInfo.phyId = in.phyId;
1411 : // urma_create_context(eidIndex) 决定 ctx 的 local EID，须与 GetTpList/Import 使用的链路 EID 一致
1412 11 : ctxInfo.ub.eidIndex = 0U;
1413 11 : const Eid linkEid = in.addr.GetEid();
1414 : try {
1415 11 : const vector<HrtDevEidInfo> eidInfoList = HrtRaGetDevEidInfoList(HRaInfo(in.mode, in.phyId));
1416 11 : bool matched = false;
1417 11 : for (const auto& eidInfo : eidInfoList) {
1418 0 : if (eidInfo.ipAddress.GetEid() == linkEid) {
1419 0 : ctxInfo.ub.eidIndex = eidInfo.eidIndex;
1420 0 : matched = true;
1421 0 : HCCL_INFO(
1422 : "[HrtRaUbCtxInit] linkEid[%s] matched eidIndex[%u].", in.addr.Describe().c_str(), eidInfo.eidIndex);
1423 0 : break;
1424 : }
1425 : }
1426 11 : if (!matched) {
1427 33 : HCCL_WARNING(
1428 : "[HrtRaUbCtxInit] linkEid[%s] not found in dev eid list(size[%zu]), "
1429 : "fallback eidIndex[0].",
1430 : in.addr.Describe().c_str(), eidInfoList.size());
1431 : }
1432 11 : } catch (const NetworkApiException&) {
1433 0 : HCCL_WARNING(
1434 : "[HrtRaUbCtxInit] HrtRaGetDevEidInfoList failed, fallback eidIndex[0], addr[%s].",
1435 : in.addr.Describe().c_str());
1436 0 : }
1437 33 : HCCL_INFO("[HrtRaUbCtxInit] use eid[%s] eidIndex[%u]", in.addr.Describe().c_str(), ctxInfo.ub.eidIndex);
1438 : s32 sRet
1439 11 : = memcpy_s(ctxInfo.ub.eid.raw, sizeof(ctxInfo.ub.eid.raw), in.addr.GetEid().raw, sizeof(in.addr.GetEid().raw));
1440 11 : if (sRet != EOK) {
1441 0 : MACRO_THROW(InternalException, StringFormat("[HrtRaUbCtxInit]memcpy_s failed. sRet[%d]", sRet));
1442 : }
1443 :
1444 : RdmaHandle handle;
1445 11 : s32 ret = RaCtxInit(&initCfg, &ctxInfo, &handle);
1446 11 : if (ret != 0) {
1447 : string msg = StringFormat(
1448 : "[Init][RaUbCtx]errNo[0x%016llx] ub ctx init fail, mode[%d], phyId[%u], addr[%s], ret[%d]",
1449 0 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), in.mode, in.phyId, in.addr.GetIpStr().c_str(), ret);
1450 0 : MACRO_THROW(NetworkApiException, msg);
1451 0 : }
1452 11 : return handle;
1453 : }
1454 :
1455 28 : void HrtRaUbCtxDestroy(RdmaHandle handle)
1456 : {
1457 28 : CHECK_NULLPTR(handle, "[HrtRaUbCtxDestroy] handle is nullptr!");
1458 84 : HCCL_INFO("[HrtRaUbCtxDestroy] rdmaHandle[%llu].", handle);
1459 28 : s32 ret = RaCtxDeinit(handle);
1460 28 : if (ret != 0) {
1461 : string msg = StringFormat(
1462 : "[DeInit][RaRdma]errNo[0x%016llx] rt ctx deinit fail. handle[%p], return[%d].",
1463 0 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), handle, ret);
1464 0 : MACRO_THROW(NetworkApiException, msg);
1465 0 : }
1466 28 : }
1467 :
1468 165 : std::pair<TokenIdHandle, uint32_t> RaUbAllocTokenIdHandle(RdmaHandle handle)
1469 : {
1470 165 : CHECK_NULLPTR(handle, "[RaUbAllocTokenIdHandle] handle is nullptr!");
1471 495 : HCCL_INFO("[RaUbAllocTokenIdHandle] rdmaHandle[%p].", handle);
1472 165 : struct HccpTokenId out {};
1473 165 : void* tokenIdHandle = nullptr;
1474 165 : s32 ret = RaCtxTokenIdAlloc(handle, &out, &tokenIdHandle);
1475 165 : if (ret != 0) {
1476 0 : string msg = StringFormat("%s failed, set=%d, rdmaHandle=%p", __func__, ret, handle);
1477 0 : MACRO_THROW(NetworkApiException, msg);
1478 0 : }
1479 495 : HCCL_INFO("[RaUbAllocTokenIdHandle] tokenIdHandle[%p], rdmaHandle[%p]", tokenIdHandle, handle);
1480 165 : return {reinterpret_cast<TokenIdHandle>(tokenIdHandle), out.tokenId >> URMA_TOKEN_ID_RIGHT_SHIFT};
1481 : }
1482 :
1483 7 : void RaUbFreeTokenIdHandle(RdmaHandle handle, TokenIdHandle tokenIdHandle)
1484 : {
1485 7 : CHECK_NULLPTR(handle, "[RaUbFreeTokenIdHandle] handle is nullptr!");
1486 21 : HCCL_INFO("[RaUbFreeTokenIdHandle] rdmaHandle[%p], tokenIdHandle[0x%llx].", handle, tokenIdHandle);
1487 7 : s32 ret = RaCtxTokenIdFree(handle, reinterpret_cast<void*>(tokenIdHandle));
1488 7 : if (ret != 0) {
1489 : string msg = StringFormat(
1490 1 : "%s failed, set=%d, rdmaHandle=%p, tokenIdHandle=0x%llx.", __func__, ret, handle, tokenIdHandle);
1491 4 : MACRO_THROW(NetworkApiException, msg);
1492 1 : }
1493 6 : }
1494 :
1495 : constexpr u64 UB_MEM_PAGE_SIZE = 4096;
1496 :
1497 639 : std::pair<u64, u64> BufAlign(u64 addr, u64 size)
1498 : {
1499 1917 : HCCL_INFO("[BufAlign] Input params: addr=0x%llx, size=%llu", addr, size);
1500 : // 待解决: 正式方案待讨论
1501 639 : u64 pageSize = UB_MEM_PAGE_SIZE;
1502 639 : u64 newAddr = addr & (~(static_cast<u64>(pageSize - 1))); // UB内存注册要求起始地址4k对齐
1503 639 : u64 offset = addr - newAddr;
1504 639 : u64 newSize = size + offset;
1505 1917 : HCCL_INFO("UB mem info: newAddr[%llx], newSize[%llu]", newAddr, newSize);
1506 :
1507 1278 : return std::make_pair(newAddr, newSize);
1508 : }
1509 :
1510 616 : HrtRaUbLocalMemRegOutParam HrtRaUbLocalMemReg(RdmaHandle handle, const HrtRaUbLocMemRegParam& in)
1511 : {
1512 616 : CHECK_NULLPTR(handle, "[HrtRaUbLocalMemReg] handle is nullptr!");
1513 1848 : HCCL_INFO("[HrtRaUbLocalMemReg] Input params: handle=%p, addr=0x%llx, size=%llu", handle, in.addr, in.size);
1514 616 : struct MrRegInfoT info {};
1515 616 : info.in.mem.addr = in.addr;
1516 616 : info.in.mem.size = in.size;
1517 :
1518 616 : info.in.ub.flags.value = 0;
1519 616 : info.in.ub.flags.bs.tokenPolicy = TOKEN_POLICY_PLAIN_TEXT;
1520 616 : info.in.ub.flags.bs.tokenIdValid = 1;
1521 616 : info.in.ub.flags.bs.access = MEM_SEG_ACCESS_READ | MEM_SEG_ACCESS_WRITE | MEM_SEG_ACCESS_ATOMIC;
1522 616 : info.in.ub.flags.bs.nonPin = in.nonPin;
1523 616 : info.in.ub.tokenValue = in.tokenValue;
1524 616 : info.in.ub.tokenIdHandle = reinterpret_cast<void*>(in.tokenIdHandle);
1525 :
1526 616 : void* lmemHandle = nullptr;
1527 616 : s32 ret = RaCtxLmemRegister(handle, &info, &lmemHandle);
1528 616 : if (ret != 0) {
1529 0 : string msg = StringFormat("localMemReg failed, addr=0x%llx, size=0x%llx", in.addr, in.size);
1530 0 : MACRO_THROW(NetworkApiException, msg);
1531 0 : }
1532 :
1533 616 : HrtRaUbLocalMemRegOutParam out;
1534 616 : s32 sRet = memcpy_s(out.key, sizeof(out.key), info.out.key.value, info.out.key.size);
1535 616 : if (sRet != EOK) {
1536 0 : MACRO_THROW(InternalException, StringFormat("[HrtRaUbLocalMemReg]memcpy_s failed. sRet[%d]", sRet));
1537 : }
1538 :
1539 1848 : HCCL_INFO("[HrtRaUbLocalMemReg]UbLocalMemReg key.size=%u", info.out.key.size);
1540 616 : out.keySize = info.out.key.size;
1541 616 : out.handle = reinterpret_cast<LocMemHandle>(lmemHandle);
1542 616 : out.targetSegVa = info.out.ub.targetSegHandle;
1543 616 : info.in.ub.tokenValue = 0;
1544 1848 : HCCL_INFO(
1545 : "[HrtRaUbLocalMemReg]UB mem reg info: in.addr[%llx], in.size[%llu], out.targetSegVa[%llu]", in.addr, in.size,
1546 : out.targetSegVa);
1547 1232 : return out;
1548 : }
1549 :
1550 1 : void HrtRaUbLocalMemUnreg(RdmaHandle rdmaHandle, LocMemHandle lmemHandle)
1551 : {
1552 1 : CHECK_NULLPTR(rdmaHandle, "[HrtRaUbLocalMemUnreg] rdmaHandle is nullptr!");
1553 3 : HCCL_INFO("[HrtRaUbLocalMemUnreg] Input params: rdmaHandle=%p, lmemHandle=0x%llx", rdmaHandle, lmemHandle);
1554 1 : s32 ret = RaCtxLmemUnregister(rdmaHandle, reinterpret_cast<void*>(lmemHandle));
1555 1 : if (ret != 0) {
1556 0 : string msg = StringFormat("localMemUnreg failed, rdmaHandle=%p, lmemHandle=0x%llx", rdmaHandle, lmemHandle);
1557 0 : MACRO_THROW(NetworkApiException, msg);
1558 0 : }
1559 1 : }
1560 :
1561 1 : HrtRaUbRemMemImportedOutParam HrtRaUbRemoteMemImport(RdmaHandle handle, u8* key, u32 keyLen, u32 tokenValue)
1562 : {
1563 2 : CHECK_NULLPTR(handle, "[HrtRaUbRemoteMemImport] handle is nullptr!");
1564 1 : CHECK_NULLPTR(key, "[HrtRaUbRemoteMemImport] key is nullptr!");
1565 3 : HCCL_INFO("[HrtRaUbRemoteMemImport] Input params: handle=%p, key=%d, keyLen=%u", handle, *key, keyLen);
1566 1 : struct MrImportInfoT info {};
1567 1 : int res = memcpy_s(info.in.key.value, sizeof(info.in.key.value), key, keyLen);
1568 1 : if (res != 0) {
1569 0 : MACRO_THROW(
1570 : InternalException, StringFormat(
1571 : "[%s] memcpy_s failed, ret = %d, params: handle=%p, key=%d, keyLen=%u", __func__,
1572 : res, handle, *key, keyLen));
1573 : }
1574 1 : info.in.key.size = keyLen;
1575 :
1576 1 : info.in.ub.tokenValue = tokenValue;
1577 1 : info.in.ub.mappingAddr = 0;
1578 1 : info.in.ub.flags.value = 0;
1579 1 : info.in.ub.flags.bs.access = MEM_SEG_ACCESS_READ | MEM_SEG_ACCESS_WRITE | MEM_SEG_ACCESS_ATOMIC;
1580 :
1581 1 : void* rmemHandle = nullptr;
1582 1 : s32 ret = RaCtxRmemImport(handle, &info, &rmemHandle);
1583 1 : if (ret != 0) {
1584 0 : string msg = StringFormat("ubRemoteMemImport failed!");
1585 0 : MACRO_THROW(NetworkApiException, msg);
1586 0 : }
1587 :
1588 1 : HrtRaUbRemMemImportedOutParam out;
1589 1 : out.handle = reinterpret_cast<LocMemHandle>(rmemHandle);
1590 1 : out.targetSegVa = info.out.ub.targetSegHandle;
1591 1 : info.in.ub.tokenValue = 0;
1592 1 : return out;
1593 : }
1594 1 : void HrtRaUbRemoteMemUnimport(RdmaHandle rdmaHandle, RemMemHandle rmemHandle)
1595 : {
1596 1 : CHECK_NULLPTR(rdmaHandle, "[HrtRaUbRemoteMemUnimport] rdmaHandle is nullptr!");
1597 3 : HCCL_INFO("[HrtRaUbRemoteMemUnimport] Input params: rdmaHandle=%p, rmemHandle=0x%llx", rdmaHandle, rmemHandle);
1598 1 : s32 ret = RaCtxRmemUnimport(rdmaHandle, reinterpret_cast<void*>(rmemHandle));
1599 1 : if (ret != 0) {
1600 : string msg
1601 0 : = StringFormat("ubRemoteMemUnimport failed, rdmaHandle=%p, rmemHandle=0x%llx", rdmaHandle, rmemHandle);
1602 0 : MACRO_THROW(NetworkApiException, msg);
1603 0 : }
1604 1 : }
1605 :
1606 : const std::map<HrtUbJfcMode, JfcMode> HRT_UB_JFC_MODE_MAP
1607 : = {{HrtUbJfcMode::NORMAL, JfcMode::JFC_MODE_NORMAL},
1608 : {HrtUbJfcMode::STARS_POLL, JfcMode::JFC_MODE_STARS_POLL},
1609 : {HrtUbJfcMode::CCU_POLL, JfcMode::JFC_MODE_CCU_POLL},
1610 : {HrtUbJfcMode::USER_CTL, JfcMode::JFC_MODE_USER_CTL_NORMAL}};
1611 :
1612 : constexpr u32 CQ_DEPTH = 2 * 1024 * 1024 / 64;
1613 : constexpr u32 CCU_CQ_DEPTH = 64;
1614 :
1615 7 : JfcHandle HrtRaUbCreateJfc(RdmaHandle handle, CqCreateInfo& cqInfo, HrtUbJfcMode mode)
1616 : {
1617 7 : CHECK_NULLPTR(handle, "[HrtRaUbCreateJfc] handle is nullptr!");
1618 21 : HCCL_INFO("[HrtRaUbCreateJfc] Input params: handle=%p, mode=%d", handle, mode);
1619 7 : struct CqInfoT info {};
1620 :
1621 7 : info.in.chanHandle = nullptr;
1622 7 : if (mode == HrtUbJfcMode::CCU_POLL) {
1623 2 : info.in.depth = CCU_CQ_DEPTH;
1624 : } else {
1625 5 : info.in.depth = CQ_DEPTH;
1626 : }
1627 7 : info.in.ub.userCtx = 0;
1628 7 : info.in.ub.mode = HRT_UB_JFC_MODE_MAP.at(mode);
1629 7 : info.in.ub.ceqn = 0;
1630 7 : info.in.ub.flag.value = 0;
1631 :
1632 7 : void* jfcHandle = nullptr;
1633 :
1634 7 : s32 ret = RaCtxCqCreate(handle, &info, &jfcHandle);
1635 7 : if (ret != 0) {
1636 0 : string msg = StringFormat("ubCreateCq failed, rdmaHandle=%p,", handle);
1637 0 : MACRO_THROW(NetworkApiException, msg);
1638 0 : }
1639 :
1640 7 : cqInfo.va = info.out.va;
1641 21 : HCCL_INFO("HrtRaUbCreateJfc va[%llu] mode[%u] jfcHandle[%p]", cqInfo.va, info.in.ub.mode, jfcHandle);
1642 7 : return reinterpret_cast<JfcHandle>(jfcHandle);
1643 : }
1644 :
1645 5 : void HrtRaUbDestroyJfc(RdmaHandle handle, JfcHandle jfcHandle)
1646 : {
1647 6 : CHECK_NULLPTR(handle, "[HrtRaUbDestroyJfc] handle is nullptr!");
1648 12 : HCCL_INFO("[HrtRaUbDestroyJfc] Input params: handle=%p, jfcHandle=0x%llx", handle, jfcHandle);
1649 4 : s32 ret = RaCtxCqDestroy(handle, reinterpret_cast<void*>(jfcHandle));
1650 4 : if (ret != 0) {
1651 0 : string msg = StringFormat("ubCqDestroy failed, rdmaHandle=%p, jfcHandle=0x%llx", handle, jfcHandle);
1652 0 : MACRO_THROW(NetworkApiException, msg);
1653 0 : }
1654 4 : }
1655 :
1656 1 : JfcHandle HrtRaUbCreateJfcUserCtl(RdmaHandle handle, CqCreateInfo& cqInfo)
1657 : {
1658 1 : CHECK_NULLPTR(handle, "[HrtRaUbCreateJfcUserCtl] handle is nullptr!");
1659 3 : HCCL_INFO("[HrtRaUbCreateJfcUserCtl] Input params: handle=%p", handle);
1660 1 : struct CqInfoT info {};
1661 :
1662 1 : info.in.chanHandle = nullptr;
1663 1 : info.in.depth = CQ_DEPTH;
1664 1 : info.in.ub.userCtx = 0;
1665 1 : info.in.ub.mode = JfcMode::JFC_MODE_USER_CTL_NORMAL;
1666 1 : info.in.ub.ceqn = 0;
1667 1 : info.in.ub.flag.value = 0;
1668 :
1669 1 : void* jfcHandle = nullptr;
1670 :
1671 1 : s32 ret = RaCtxCqCreate(handle, &info, &jfcHandle);
1672 1 : if (ret != 0) {
1673 0 : string msg = StringFormat("ubCreateCq failed, rdmaHandle=%p,", handle);
1674 0 : THROW<NetworkApiException>(msg);
1675 0 : }
1676 :
1677 3 : HCCL_INFO(
1678 : "[HrtRaUbCreateJfcUserCtl] jfcId[%u], cqVA[%llx], cqeSize[%u], cqDepth[%u], dbAddr[%llx]", info.out.id,
1679 : info.out.bufAddr, info.out.cqeSize, CQ_DEPTH, info.out.swdbAddr);
1680 :
1681 1 : cqInfo.va = info.out.bufAddr;
1682 1 : cqInfo.id = info.out.id;
1683 1 : cqInfo.cqeSize = info.out.cqeSize;
1684 1 : cqInfo.cqDepth = CQ_DEPTH;
1685 1 : cqInfo.swdbAddr = info.out.swdbAddr;
1686 :
1687 1 : return reinterpret_cast<JfcHandle>(jfcHandle);
1688 : }
1689 :
1690 : const std::map<HrtTransportMode, TransportModeT> HRT_TRANSPORT_MODE_MAP
1691 : = {{HrtTransportMode::RM, TransportModeT::CONN_RM}};
1692 :
1693 : const std::map<HrtJettyMode, JettyMode> HRT_JETTY_MODE_MAP
1694 : = {{HrtJettyMode::STANDARD, JettyMode::JETTY_MODE_URMA_NORMAL},
1695 : {HrtJettyMode::HOST_OFFLOAD, JettyMode::JETTY_MODE_USER_CTL_NORMAL},
1696 : {HrtJettyMode::HOST_OPBASE, JettyMode::JETTY_MODE_USER_CTL_NORMAL},
1697 : {HrtJettyMode::DEV_USED, JettyMode::JETTY_MODE_USER_CTL_NORMAL},
1698 : {HrtJettyMode::CACHE_LOCK_DWQE, JettyMode::JETTY_MODE_CACHE_LOCK_DWQE},
1699 : {HrtJettyMode::CCU_CCUM_CACHE, JettyMode::JETTY_MODE_CCU}};
1700 :
1701 : constexpr u8 RNR_RETRY = 7;
1702 : constexpr u32 RQ_DEPTH = 256;
1703 :
1704 139 : static struct QpCreateAttr GetQpCreateAttr(const HrtRaUbCreateJettyParam& in)
1705 : {
1706 139 : struct QpCreateAttr attr {};
1707 139 : attr.scqHandle = reinterpret_cast<void*>(in.sjfcHandle);
1708 139 : attr.rcqHandle = reinterpret_cast<void*>(in.rjfcHandle);
1709 139 : attr.srqHandle = reinterpret_cast<void*>(in.sjfcHandle);
1710 139 : attr.rqDepth = RQ_DEPTH;
1711 139 : attr.sqDepth = in.sqDepth;
1712 139 : attr.transportMode = HRT_TRANSPORT_MODE_MAP.at(in.transMode);
1713 139 : attr.ub.mode = HRT_JETTY_MODE_MAP.at(in.jettyMode);
1714 :
1715 139 : attr.ub.tokenValue = in.tokenValue;
1716 139 : attr.ub.tokenIdHandle = reinterpret_cast<void*>(in.tokenIdHandle);
1717 139 : attr.ub.flag.value = 0;
1718 : /* errTime配置值：0-31
1719 : 0-7代表芯片配置值b00:512ms
1720 : 8-15代表芯片配置值b01:1s
1721 : 16-23代表芯片配置值b10:8s
1722 : 24-31代表芯片配置值b11:32s
1723 : */
1724 139 : attr.ub.errTimeout = in.errTimeout;
1725 139 : attr.ub.priority = static_cast<uint8_t>(in.qos & 0xFU);
1726 139 : attr.ub.rnrRetry = RNR_RETRY;
1727 139 : attr.ub.flag.bs.shareJfr = 1;
1728 139 : attr.ub.jettyId = in.jettyId;
1729 : // 在continue模式下+配置了wqe的fence标记，并且远端有一些权限校验错误/内存异常错误，硬件会直接挂死
1730 : // jfs_flag 的 error_suspend 设置为 1，
1731 139 : attr.ub.jfsFlag.bs.errorSuspend = 1;
1732 :
1733 139 : attr.ub.extMode.sqebbNum = in.sqDepth;
1734 139 : if (in.jettyMode == HrtJettyMode::HOST_OFFLOAD) {
1735 6 : attr.ub.extMode.piType = 1;
1736 6 : attr.ub.extMode.cstmFlag.bs.sqCstm = 0; // 表示不指定Va，由HCCP返回Va
1737 133 : } else if (in.jettyMode == HrtJettyMode::CCU_CCUM_CACHE) {
1738 26 : attr.ub.tokenValue = in.tokenValue;
1739 26 : attr.ub.extMode.cstmFlag.bs.sqCstm = 1;
1740 26 : attr.ub.extMode.sq.buffSize = in.sqBufSize;
1741 26 : attr.ub.extMode.sq.buffVa = in.sqBufVa;
1742 107 : } else if (in.jettyMode == HrtJettyMode::DEV_USED || in.jettyMode == HrtJettyMode::CACHE_LOCK_DWQE) {
1743 1 : attr.ub.extMode.cstmFlag.bs.sqCstm = 0; // 表示不指定Va，由HCCP返回Va
1744 1 : attr.ub.extMode.sq.buffSize = in.sqBufSize;
1745 1 : attr.ub.extMode.sq.buffVa = in.sqBufVa;
1746 : } // 预埋HrtJettyMode::CACHE_LOCK_DWQE类型，当前流程暂未使用
1747 :
1748 : // 其他Mode暂时不需要额外更新特定字段
1749 417 : HCCL_INFO(
1750 : "Create jetty, input params: attr.ub.jettyId[%u], attr.rqDepth[%u], "
1751 : "attr.sqDepth[%u], attr.transportMode[%d], attr.ub.mode[%d], "
1752 : "attr.ub.extMode.sqebbNum[%u], attr.ub.extMode.sq.buffVa[%llx], "
1753 : "attr.ub.extMode.sq.buffSize[%u], attr.ub.extMode.piType[%u], attr.ub.priority[%u], timeout[%u].",
1754 : attr.ub.jettyId, attr.rqDepth, attr.sqDepth, attr.transportMode, attr.ub.mode, attr.ub.extMode.sqebbNum,
1755 : attr.ub.extMode.sq.buffVa, attr.ub.extMode.sq.buffSize, attr.ub.extMode.piType, attr.ub.priority,
1756 : attr.ub.errTimeout);
1757 139 : return attr;
1758 : }
1759 :
1760 29 : HrtRaUbJettyCreatedOutParam HrtRaUbCreateJetty(RdmaHandle handle, const HrtRaUbCreateJettyParam& in)
1761 : {
1762 29 : CHECK_NULLPTR(handle, "[HrtRaUbCreateJetty] handle is nullptr!");
1763 87 : HCCL_INFO("[HrtRaUbCreateJetty] Input params: handle=%p", handle);
1764 29 : struct QpCreateAttr attr = GetQpCreateAttr(in);
1765 :
1766 29 : struct QpCreateInfo info {};
1767 29 : void* qpHandle = nullptr;
1768 29 : s32 ret = RaCtxQpCreate(handle, &attr, &info, &qpHandle);
1769 29 : if (ret != 0) {
1770 0 : string msg = StringFormat("ubCreateJetty failed, rdmaHandle=%p,", handle);
1771 0 : MACRO_THROW(NetworkApiException, msg);
1772 0 : }
1773 :
1774 29 : HrtRaUbJettyCreatedOutParam out;
1775 29 : out.handle = reinterpret_cast<JettyHandle>(qpHandle);
1776 29 : out.id = info.ub.id;
1777 29 : out.uasid = info.ub.uasid;
1778 29 : out.jettyVa = info.va;
1779 29 : out.dbVa = info.ub.dbAddr;
1780 29 : out.dbTokenId = info.ub.dbTokenId >> URMA_TOKEN_ID_RIGHT_SHIFT;
1781 29 : out.sqBuffVa = info.ub.sqBuffVa; // 适配HCCP修改，jettybufva由HCCP提供，不再由HCCL分配
1782 :
1783 29 : s32 sRet = memcpy_s(out.key, sizeof(out.key), info.key.value, info.key.size);
1784 29 : if (sRet != EOK) {
1785 0 : MACRO_THROW(
1786 : InternalException,
1787 : StringFormat("HrtRaUbCreateJetty memcpy_s failed. sRet[%d], params: handle=%p", sRet, handle));
1788 : }
1789 29 : out.keySize = info.key.size;
1790 29 : attr.ub.tokenValue = 0;
1791 87 : HCCL_INFO("Create jetty success, handle[%llu] jettyVa[%llu]", out.handle, out.jettyVa);
1792 58 : return out;
1793 : }
1794 :
1795 1 : void HrtRaUbDestroyJetty(JettyHandle jettyHandle)
1796 : {
1797 3 : HCCL_INFO("[HrtRaUbDestroyJetty] Input params: jettyHandle=0x%llx", jettyHandle);
1798 1 : s32 ret = RaCtxQpDestroy(reinterpret_cast<void*>(jettyHandle));
1799 1 : if (ret != 0) {
1800 0 : string msg = StringFormat("ubDestroyJetty failed, jettyHandle=0x%llx", jettyHandle);
1801 0 : MACRO_THROW(NetworkApiException, msg);
1802 0 : }
1803 1 : }
1804 :
1805 25 : static HrtRaUbJettyImportedOutParam ImportJetty(
1806 : RdmaHandle handle, u8* key, u32 keyLen, u32 tokenValue, JettyImportExpCfg cfg, JettyImportMode mode,
1807 : TpProtocol protocol = TpProtocol::INVALID)
1808 : {
1809 50 : CHECK_NULLPTR(handle, "[ImportJetty] handle is nullptr!");
1810 25 : CHECK_NULLPTR(key, "[ImportJetty] key is nullptr!");
1811 75 : HCCL_INFO("[ImportJetty] Input params: handle=%p, key=%d, keyLen=%u, mode=%d", handle, *key, keyLen, mode);
1812 25 : if (mode == JettyImportMode::JETTY_IMPORT_MODE_NORMAL) {
1813 0 : MACRO_THROW(
1814 : NotSupportException, StringFormat("[%s] currently not support JETTY_IMPORT_MODE_NORMAL.", __func__));
1815 : }
1816 :
1817 25 : struct QpImportInfoT info {};
1818 :
1819 25 : int res = memcpy_s(info.in.key.value, sizeof(info.in.key.value), key, keyLen);
1820 25 : if (res != 0) {
1821 0 : MACRO_THROW(InternalException, StringFormat("[%s] memcpy_s failed, ret = %d", __func__, res));
1822 : }
1823 25 : info.in.key.size = keyLen;
1824 :
1825 25 : info.in.ub.mode = mode;
1826 25 : info.in.ub.tokenValue = tokenValue;
1827 25 : info.in.ub.policy = JettyGrpPolicy::JETTY_GRP_POLICY_RR;
1828 25 : info.in.ub.type = TargetType::TARGET_TYPE_JETTY;
1829 :
1830 25 : info.in.ub.flag.value = 0;
1831 25 : info.in.ub.flag.bs.tokenPolicy = TOKEN_POLICY_PLAIN_TEXT;
1832 :
1833 25 : info.in.ub.expImportCfg = cfg;
1834 :
1835 26 : if (protocol != TpProtocol::TP && protocol != TpProtocol::CTP && protocol != TpProtocol::UBOE
1836 26 : && protocol != TpProtocol::UBG) {
1837 4 : MACRO_THROW(
1838 : NetworkApiException,
1839 : StringFormat("[%s] failed, tp protocol[%s] is not expected.", __func__, protocol.Describe().c_str()));
1840 : }
1841 : // tpType: 0->RTP, 1->CTP
1842 24 : info.in.ub.tpType = protocol == TpProtocol::TP ? 0 : 1;
1843 :
1844 24 : void* remQpHandle = nullptr;
1845 24 : s32 ret = RaCtxQpImport(handle, &info, &remQpHandle);
1846 24 : if (ret != 0) {
1847 0 : string msg = StringFormat("UbImportJetty failed, rdmaHandle=%p,", handle);
1848 0 : MACRO_THROW(NetworkApiException, msg);
1849 0 : }
1850 :
1851 24 : HrtRaUbJettyImportedOutParam out;
1852 24 : out.handle = reinterpret_cast<TargetJettyHandle>(remQpHandle);
1853 24 : out.targetJettyVa = info.out.ub.tjettyHandle;
1854 24 : out.tpn = info.out.ub.tpn;
1855 :
1856 72 : HCCL_INFO("ImportJetty handle[%llu] targetJettyVa[%llu] tpn[%u]", out.handle, out.targetJettyVa, out.tpn);
1857 24 : info.in.ub.tokenValue = 0;
1858 48 : return out;
1859 : }
1860 :
1861 42 : static struct JettyImportExpCfg GetTpImportCfg(const JettyImportCfg& jettyImportCfg)
1862 : {
1863 42 : struct JettyImportExpCfg cfg = {};
1864 :
1865 42 : cfg.tpHandle = jettyImportCfg.localTpHandle;
1866 42 : cfg.peerTpHandle = jettyImportCfg.remoteTpHandle;
1867 42 : cfg.tag = jettyImportCfg.localTag;
1868 42 : cfg.txPsn = jettyImportCfg.localPsn;
1869 42 : cfg.rxPsn = jettyImportCfg.remotePsn;
1870 :
1871 126 : HCCL_INFO(
1872 : "GetTpImportCfg tpHandle[%llu] peerTpHandle[%llu] tag[%llu] txPsn[%llu] rxPsn[%llu]", cfg.tpHandle,
1873 : cfg.peerTpHandle, cfg.tag, cfg.txPsn, cfg.rxPsn);
1874 :
1875 42 : return cfg;
1876 : }
1877 :
1878 0 : HrtRaUbJettyImportedOutParam RaUbImportJetty(RdmaHandle handle, u8* key, u32 keyLen, u32 tokenValue)
1879 : {
1880 0 : CHECK_NULLPTR(handle, "[RaUbImportJetty] handle is nullptr!");
1881 0 : CHECK_NULLPTR(key, "[RaUbImportJetty] key is nullptr!");
1882 0 : HCCL_INFO("[RaUbImportJetty] Input params: handle=%p, key=%d, keyLen=%u", handle, *key, keyLen);
1883 : // 该接口仅适配非管控面模式，当前不期望使用
1884 0 : struct JettyImportExpCfg cfg = {};
1885 0 : const auto mode = JettyImportMode::JETTY_IMPORT_MODE_NORMAL;
1886 0 : return ImportJetty(handle, key, keyLen, tokenValue, cfg, mode);
1887 : }
1888 :
1889 : HrtRaUbJettyImportedOutParam
1890 25 : RaUbTpImportJetty(RdmaHandle handle, u8* key, u32 keyLen, u32 tokenValue, const JettyImportCfg& jettyImportCfg)
1891 : {
1892 50 : CHECK_NULLPTR(handle, "[RaUbTpImportJetty] handle is nullptr!");
1893 25 : CHECK_NULLPTR(key, "[RaUbTpImportJetty] key is nullptr!");
1894 75 : HCCL_INFO("[RaUbTpImportJetty] Input params: handle=%p", handle);
1895 25 : struct JettyImportExpCfg cfg = GetTpImportCfg(jettyImportCfg);
1896 25 : const auto mode = JettyImportMode::JETTY_IMPORT_MODE_EXP;
1897 49 : return ImportJetty(handle, key, keyLen, tokenValue, cfg, mode, jettyImportCfg.protocol);
1898 : }
1899 :
1900 1 : void HrtRaUbUnimportJetty(RdmaHandle handle, TargetJettyHandle targetJettyHandle)
1901 : {
1902 1 : CHECK_NULLPTR(handle, "[HrtRaUbUnimportJetty] handle is nullptr!");
1903 3 : HCCL_INFO("[HrtRaUbUnimportJetty] Input params: handle=%p, targetJettyHandle=0x%llx", handle, targetJettyHandle);
1904 1 : s32 ret = RaCtxQpUnimport(reinterpret_cast<void*>(handle), reinterpret_cast<void*>(targetJettyHandle));
1905 1 : if (ret != 0) {
1906 : string msg
1907 0 : = StringFormat("ubCqDestroy failed, rdmaHandle=%p, targetJettyHandle=0x%llx", handle, targetJettyHandle);
1908 0 : MACRO_THROW(NetworkApiException, msg);
1909 0 : }
1910 1 : }
1911 :
1912 1 : void HrtRaUbJettyBind(JettyHandle jettyHandle, TargetJettyHandle targetJettyHandle)
1913 : {
1914 3 : HCCL_INFO(
1915 : "[HrtRaUbJettyBind] Input params: jettyHandle=0x%llx, targetJettyHandle=0x%llx", jettyHandle,
1916 : targetJettyHandle);
1917 1 : s32 ret = RaCtxQpBind(reinterpret_cast<void*>(jettyHandle), reinterpret_cast<void*>(targetJettyHandle));
1918 1 : if (ret != 0) {
1919 : string msg = StringFormat(
1920 0 : "ubJettyBind failed, jettyHandle=0x%llx, targetJettyHandle=0x%llx", jettyHandle, targetJettyHandle);
1921 0 : MACRO_THROW(NetworkApiException, msg);
1922 0 : }
1923 1 : }
1924 :
1925 1 : void HrtRaUbJettyUnbind(JettyHandle jettyHandle)
1926 : {
1927 3 : HCCL_INFO("[HrtRaUbJettyUnbind] Input params: jettyHandle=0x%llx", jettyHandle);
1928 1 : s32 ret = RaCtxQpUnbind(reinterpret_cast<void*>(jettyHandle));
1929 1 : if (ret != 0) {
1930 0 : string msg = StringFormat("ubJettyUnbind failed, jettyHandle=0x%llx", jettyHandle);
1931 0 : MACRO_THROW(NetworkApiException, msg);
1932 0 : }
1933 1 : }
1934 :
1935 : const std::map<HrtUbSendWrOpCode, RaUbOpcode> HRT_UB_SEND_WR_OP_CODE_MAP
1936 : = {{HrtUbSendWrOpCode::WRITE, RaUbOpcode::RA_UB_OPC_WRITE},
1937 : {HrtUbSendWrOpCode::WRITE_WITH_NOTIFY, RaUbOpcode::RA_UB_OPC_WRITE_NOTIFY},
1938 : {HrtUbSendWrOpCode::READ, RaUbOpcode::RA_UB_OPC_READ},
1939 : {HrtUbSendWrOpCode::NOP, RaUbOpcode::RA_UB_OPC_NOP}};
1940 :
1941 : const std::map<ReduceOp, u8> HRT_UB_REDUCE_OP_CODE_MAP
1942 : = {{ReduceOp::SUM, 0xA}, {ReduceOp::MAX, 0x8}, {ReduceOp::MIN, 0x9}};
1943 :
1944 : const std::map<DataType, u8> HRT_UB_REDUCE_DATA_TYPE_MAP
1945 : = {{DataType::INT8, 0x0}, {DataType::INT16, 0x1}, {DataType::INT32, 0x2}, {DataType::UINT8, 0x3},
1946 : {DataType::UINT16, 0x4}, {DataType::UINT32, 0x5}, {DataType::FP16, 0x6}, {DataType::FP32, 0x7},
1947 : {DataType::BFP16, 0x8}, {DataType::BF16_SAT, 0x9}};
1948 :
1949 9 : static void ConstructWrSge(HrtRaUbSendWrReqParam& in, struct WrSgeList& sge)
1950 : {
1951 9 : sge.addr = in.localAddr;
1952 9 : sge.len = in.size;
1953 9 : sge.lmemHandle = reinterpret_cast<void*>(in.lmemHandle);
1954 9 : }
1955 :
1956 9 : static void ConstructSendWrReq(HrtRaUbSendWrReqParam& in, struct WrSgeList& sge, struct SendWrData& sendWr)
1957 : {
1958 : // 看一下hccp测试用例的入参
1959 9 : sendWr.numSge = 1;
1960 9 : sendWr.sges = &sge;
1961 9 : sendWr.remoteAddr = in.remoteAddr;
1962 9 : sendWr.rmemHandle = reinterpret_cast<void*>(in.rmemHandle);
1963 9 : sendWr.ub.userCtx = 0;
1964 9 : sendWr.ub.opcode = HRT_UB_SEND_WR_OP_CODE_MAP.at(in.opcode);
1965 9 : sendWr.ub.flags.value = 0;
1966 9 : sendWr.ub.flags.bs.compOrder = 1;
1967 9 : sendWr.ub.flags.bs.completeEnable = in.cqeEn;
1968 9 : sendWr.ub.flags.bs.fence = 1;
1969 9 : sendWr.ub.flags.bs.solicitedEnable = 1;
1970 9 : sendWr.ub.remQpHandle = reinterpret_cast<void*>(in.handle);
1971 9 : sendWr.ub.flags.bs.inlineFlag = in.inlineFlag;
1972 9 : if (sendWr.ub.flags.bs.inlineFlag) {
1973 3 : sendWr.inlineData = in.inlineData;
1974 3 : sendWr.inlineSize = in.size;
1975 : }
1976 9 : sendWr.ub.reduceInfo.reduceEn = in.inlineReduceFlag;
1977 9 : if (sendWr.ub.reduceInfo.reduceEn) {
1978 4 : sendWr.ub.reduceInfo.reduceOpcode = HRT_UB_REDUCE_OP_CODE_MAP.at(in.reduceOp);
1979 4 : sendWr.ub.reduceInfo.reduceDataType = HRT_UB_REDUCE_DATA_TYPE_MAP.at(in.dataType);
1980 : }
1981 9 : if (sendWr.ub.opcode == RaUbOpcode::RA_UB_OPC_WRITE_NOTIFY) {
1982 3 : sendWr.ub.notifyInfo.notifyData = in.notifyData;
1983 3 : sendWr.ub.notifyInfo.notifyAddr = in.notifyAddr;
1984 3 : sendWr.ub.notifyInfo.notifyHandle = reinterpret_cast<void*>(in.notifyHandle);
1985 : }
1986 9 : }
1987 :
1988 9 : HrtRaUbSendWrRespParam HrtRaUbPostSend(JettyHandle jettyHandle, HrtRaUbSendWrReqParam& in)
1989 : {
1990 9 : struct WrSgeList sge = {};
1991 9 : struct SendWrData sendWr {};
1992 :
1993 9 : ConstructWrSge(in, sge);
1994 9 : ConstructSendWrReq(in, sge, sendWr);
1995 :
1996 27 : HCCL_INFO("Sge addr = 0x%llx", in.localAddr);
1997 27 : HCCL_INFO("SendWR lmemHandle = 0x%llx", in.lmemHandle); // 和notifyFixedValue能否对齐
1998 27 : HCCL_INFO("SendWR rmemHandle = 0x%llx", in.rmemHandle); // remote
1999 27 : HCCL_INFO("SendWR remote addr = 0x%llx", in.remoteAddr);
2000 27 : HCCL_INFO("SendWR remote qp handle = 0x%llx", in.handle);
2001 27 : HCCL_INFO("SendWR jetty handle = 0x%llx", jettyHandle);
2002 :
2003 9 : SendWrResp sendWrResp{};
2004 :
2005 9 : u32 compNum = 0;
2006 9 : s32 ret = RaBatchSendWr(reinterpret_cast<void*>(jettyHandle), &sendWr, &sendWrResp, 1, &compNum);
2007 9 : if (ret != 0) {
2008 0 : string msg = StringFormat("UbJettySendWr failed, jettyHandle=0x%llx,", jettyHandle);
2009 0 : MACRO_THROW(NetworkApiException, msg);
2010 0 : }
2011 9 : HrtRaUbSendWrRespParam out;
2012 9 : out.dieId = sendWrResp.doorbellInfo.dieId;
2013 9 : out.funcId = sendWrResp.doorbellInfo.funcId;
2014 9 : out.jettyId = sendWrResp.doorbellInfo.jettyId;
2015 9 : out.piVal = sendWrResp.doorbellInfo.piVal;
2016 9 : out.dwqeSize = sendWrResp.doorbellInfo.dwqeSize;
2017 9 : ret = memcpy_s(out.dwqe, sizeof(out.dwqe), sendWrResp.doorbellInfo.dwqe, out.dwqeSize);
2018 9 : if (ret != 0) {
2019 0 : string msg = StringFormat("HrtRaUbPostSend copy dwqe failed, ret=%d", ret);
2020 0 : MACRO_THROW(InternalException, msg);
2021 0 : }
2022 :
2023 18 : return out;
2024 : }
2025 :
2026 15 : std::pair<uint32_t, uint32_t> HraGetDieAndFuncId(RdmaHandle handle)
2027 : {
2028 15 : CHECK_NULLPTR(handle, "[HraGetDieAndFuncId] handle is nullptr!");
2029 45 : HCCL_INFO("[HraGetDieAndFuncId] Input params: handle=%p", handle);
2030 15 : struct DevBaseAttr out {};
2031 15 : auto ret = RaGetDevBaseAttr(handle, &out);
2032 15 : if (ret != 0) {
2033 0 : MACRO_THROW(
2034 : NetworkApiException, StringFormat("[%s] call ra_get_dev_base_attr failed, error code =%d.", __func__, ret));
2035 : }
2036 30 : return std::make_pair(out.ub.dieId, out.ub.funcId);
2037 : }
2038 :
2039 2 : bool HraGetRtpEnable(RdmaHandle handle)
2040 : {
2041 2 : struct DevBaseAttr out {};
2042 2 : auto ret = RaGetDevBaseAttr(handle, &out);
2043 2 : if (ret != 0) {
2044 0 : THROW<NetworkApiException>(StringFormat("[%s] call RaGetDevBaseAttr failed, error code =%d.", __func__, ret));
2045 : }
2046 :
2047 6 : HCCL_RUN_INFO(
2048 : "[%s] rmTpCap[%u] rcTpCap[%u] umTpCap[%u] tpFeat[%u]", __func__, out.ub.rmTpCap.value, out.ub.rcTpCap.value,
2049 : out.ub.umTpCap.value, out.ub.tpFeat.value);
2050 :
2051 18 : for (int i = 0; i < MAX_PRIORITY_CNT; i++) {
2052 17 : const CtxSlInfo& priorityInfo = out.ub.priorityInfo[i];
2053 51 : HCCL_RUN_INFO(
2054 : "[%s] priorityInfo[%d]: SL[%u] tpType[%u] rtp[%u]", __func__, i, priorityInfo.SL, priorityInfo.tpType.value,
2055 : priorityInfo.tpType.bs.rtp);
2056 17 : if (priorityInfo.tpType.bs.rtp == 1) {
2057 1 : return true;
2058 : }
2059 : }
2060 1 : return false;
2061 : }
2062 :
2063 2 : void HrtRaUbPostNops(JettyHandle jettyHandle, JettyHandle remoteJettyHandle, const u32 numNop)
2064 : {
2065 6 : HCCL_INFO(
2066 : "HrtRaUbPostNops: jettyHandle[0x%llx], remoteJettyHandle[0x%llx], numNop[%u]", jettyHandle, remoteJettyHandle,
2067 : numNop);
2068 131 : struct SendWrData sendWrList[numNop] = {};
2069 131 : for (auto& sendWr : sendWrList) {
2070 129 : sendWr.ub.opcode = HRT_UB_SEND_WR_OP_CODE_MAP.at(HrtUbSendWrOpCode::NOP);
2071 387 : HCCL_INFO("SendWR opcode = %u", static_cast<u32>(sendWr.ub.opcode));
2072 : }
2073 2 : sendWrList[numNop - 1].ub.flags.bs.completeEnable = 1;
2074 :
2075 131 : SendWrResp sendWrRespList[numNop] = {};
2076 2 : u32 compNum = 0;
2077 2 : s32 ret = RaBatchSendWr(reinterpret_cast<void*>(jettyHandle), sendWrList, sendWrRespList, numNop, &compNum);
2078 2 : if (ret != 0) {
2079 1 : string msg = StringFormat("UbJettySendWr failed, jettyHandle=0x%llx,", jettyHandle);
2080 4 : MACRO_THROW(NetworkApiException, msg);
2081 1 : }
2082 3 : }
2083 :
2084 1 : void RaUbUpdateCi(JettyHandle jettyHandle, u32 ci)
2085 : {
2086 3 : HCCL_INFO("RaUbUpdateCi: jettyHandle=0x%llx, ci=%u", jettyHandle, ci);
2087 1 : s32 ret = RaCtxUpdateCi(reinterpret_cast<void*>(jettyHandle), ci);
2088 1 : if (ret != 0) {
2089 1 : string msg = StringFormat("UbUpdateCi failed, ret=%d, jettyHandle=0x%llx, ci=%u", ret, jettyHandle, ci);
2090 4 : MACRO_THROW(NetworkApiException, msg);
2091 1 : }
2092 0 : }
2093 :
2094 67 : inline string HccpEidDesc(union HccpEid& hccpEid)
2095 : {
2096 : return StringFormat(
2097 67 : "HccpEid[%016llx:%016llx]", static_cast<unsigned long long>(be64toh(hccpEid.in6.subnetPrefix)),
2098 134 : static_cast<unsigned long long>(be64toh(hccpEid.in6.interfaceId)));
2099 : }
2100 :
2101 3 : inline IpAddress HccpEidToIpAddress(union HccpEid& hccpEid)
2102 : {
2103 3 : Eid eid{};
2104 9 : HCCL_INFO("[HccpEidToIpAddress] %s", HccpEidDesc(hccpEid).c_str());
2105 3 : s32 sRet = memcpy_s(eid.raw, sizeof(eid.raw), hccpEid.raw, sizeof(hccpEid.raw));
2106 3 : if (sRet != EOK) {
2107 0 : MACRO_THROW(InternalException, StringFormat("[HccpEidToIpAddress]memcpy_s failed. sRet[%d]", sRet));
2108 : }
2109 6 : return IpAddress(eid);
2110 : }
2111 :
2112 21 : std::vector<HrtDevEidInfo> HrtRaGetDevEidInfoList(const HRaInfo& raInfo)
2113 : {
2114 21 : std::vector<HrtDevEidInfo> hrtDevEidInfo;
2115 21 : struct RaInfo info {};
2116 21 : u32 num = 0;
2117 :
2118 21 : info.mode = HRT_NETWORK_MODE_MAP.at(raInfo.mode);
2119 21 : info.phyId = raInfo.phyId;
2120 :
2121 63 : HCCL_INFO("[HrtRaGetDevEidInfoList] Input params: mode=%d, phyId=%u", info.mode, info.phyId);
2122 21 : s32 ret = RaGetDevEidInfoNum(info, &num);
2123 21 : if (ret != 0) {
2124 1 : string msg = StringFormat("call RaGetDevEidInfoNum failed, error code =%d.", ret);
2125 4 : MACRO_THROW(NetworkApiException, msg);
2126 1 : }
2127 :
2128 24 : struct HccpDevEidInfo infoList[num] = {};
2129 20 : ret = RaGetDevEidInfoList(info, infoList, &num);
2130 20 : if (ret != 0) {
2131 1 : string msg = StringFormat("call RaGetDevEidInfoList failed, error code =%d.", ret);
2132 4 : MACRO_THROW(NetworkApiException, msg);
2133 1 : }
2134 :
2135 19 : hrtDevEidInfo.resize(num);
2136 21 : for (u32 i = 0; i < num; i++) {
2137 2 : hrtDevEidInfo[i].name = (infoList[i].name);
2138 2 : hrtDevEidInfo[i].ipAddress = HccpEidToIpAddress(infoList[i].eid);
2139 2 : hrtDevEidInfo[i].type = infoList[i].type;
2140 2 : hrtDevEidInfo[i].eidIndex = infoList[i].eidIndex;
2141 2 : hrtDevEidInfo[i].dieId = infoList[i].dieId;
2142 2 : hrtDevEidInfo[i].chipId = infoList[i].chipId;
2143 2 : hrtDevEidInfo[i].funcId = infoList[i].funcId;
2144 2 : hrtDevEidInfo[i].devFeature = infoList[i].devFeature;
2145 6 : HCCL_INFO(
2146 : "[%s] HrtDevEidInfo[%d]: name[%s], ipAddress[%s], type[%u], "
2147 : "eidIndex[%u], dieId[%u], chipId[%u], funcId[%u], devFeature[%u]",
2148 : __func__, i, hrtDevEidInfo[i].name.c_str(), hrtDevEidInfo[i].ipAddress.Describe().c_str(),
2149 : hrtDevEidInfo[i].type, hrtDevEidInfo[i].eidIndex, hrtDevEidInfo[i].dieId, hrtDevEidInfo[i].chipId,
2150 : hrtDevEidInfo[i].funcId, hrtDevEidInfo[i].devFeature);
2151 : }
2152 :
2153 19 : return hrtDevEidInfo;
2154 23 : }
2155 :
2156 79 : ReqHandleResult HrtRaGetAsyncReqResult(RequestHandle& reqHandle)
2157 : {
2158 79 : if (reqHandle == 0) {
2159 3 : HCCL_ERROR("[%s] failed, reqHandle is 0.params: reqHandle=0x%llx", __func__, reqHandle);
2160 1 : return ReqHandleResult::INVALID_PARA;
2161 : }
2162 :
2163 78 : int reqResult = 0;
2164 78 : s32 ret = RaGetAsyncReqResult(reinterpret_cast<void*>(reqHandle), &reqResult);
2165 : // 返回 OTHERS_EAGAIN 代表查询到异步任务未完成，需要重新查询，此时保留handle
2166 78 : if (ret == OTHERS_EAGAIN) {
2167 1 : return ReqHandleResult::NOT_COMPLETED;
2168 : }
2169 :
2170 : // 返回码非0代表调用查询接口失败，当前仅入参错误时触发
2171 77 : if (ret != 0) {
2172 4 : MACRO_THROW(
2173 : NetworkApiException, StringFormat(
2174 : "[%s] failed, call interface error[%d], "
2175 : "reqhandle[%llu].",
2176 : __func__, ret, reqHandle));
2177 : }
2178 :
2179 76 : RequestHandle tmpReqHandle = reqHandle;
2180 76 : reqHandle = 0;
2181 : // 返回码为 0 时，reqResult为异步任务完成结果，0代表成功，其他值代表失败
2182 : // SOCK_EAGAIN 为 socket 类执行结果，代表 socket 接口失败需要重试
2183 76 : if (reqResult == SOCK_EAGAIN) {
2184 1 : return ReqHandleResult::SOCK_E_AGAIN;
2185 : }
2186 :
2187 75 : if (reqResult != 0) {
2188 4 : MACRO_THROW(
2189 : NetworkApiException, StringFormat(
2190 : "[%s] failed, the asynchronous request "
2191 : "error[%d], reqhandle[%llu].",
2192 : __func__, reqResult, tmpReqHandle));
2193 : }
2194 :
2195 74 : return ReqHandleResult::COMPLETED;
2196 : }
2197 :
2198 6 : RequestHandle RaSocketConnectOneAsync(RaSocketConnectParam& in)
2199 : {
2200 18 : HCCL_INFO(
2201 : "[RaSocketConnectOneAsync] Input params: socketHandle=%p, remoteIp=%s, port=%u, tag=%s", in.socketHandle,
2202 : in.remoteIp.Describe().c_str(), in.port, in.tag.c_str());
2203 6 : struct SocketConnectInfoT connInfo {};
2204 6 : connInfo.socketHandle = in.socketHandle;
2205 6 : connInfo.remoteIp = IpAddressToHccpIpAddr(in.remoteIp);
2206 6 : connInfo.port = in.port;
2207 :
2208 6 : int sret = strcpy_s(connInfo.tag, sizeof(connInfo.tag), in.tag.c_str());
2209 6 : if (sret != 0) {
2210 0 : MACRO_THROW(
2211 : NetworkApiException,
2212 : StringFormat(
2213 : "[%s] copy tag[%s] to hccp tag failed, ret=%d, connInfo.tag size=%zu, in.tag size=%zu", __func__,
2214 : in.tag.c_str(), sret, sizeof(connInfo.tag), sizeof(in.tag.c_str())));
2215 : }
2216 :
2217 18 : HCCL_INFO("Socket Connect tag=[%s], remoteIp[%s]", connInfo.tag, in.remoteIp.Describe().c_str());
2218 6 : void* raReqHandle = nullptr;
2219 6 : int ret = RaSocketBatchConnectAsync(&connInfo, SOCKET_NUM_ONE, &raReqHandle);
2220 6 : if (ret != 0) {
2221 0 : MACRO_THROW(
2222 : NetworkApiException,
2223 : StringFormat(
2224 : "[BatchConnect][RaSocket]errNo[0x%016llx] ra socket batch connect fail. return[%d]",
2225 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret));
2226 : }
2227 :
2228 6 : return reinterpret_cast<RequestHandle>(raReqHandle);
2229 : }
2230 :
2231 1 : RequestHandle RaSocketCloseOneAsync(RaSocketCloseParam& in)
2232 : {
2233 3 : HCCL_INFO("[RaSocketCloseOneAsync] Input params: socketHandle=%p, fdHandle=%p", in.socketHandle, in.fdHandle);
2234 1 : struct SocketCloseInfoT closeInfo = {};
2235 1 : closeInfo.fdHandle = in.fdHandle;
2236 1 : closeInfo.socketHandle = in.socketHandle;
2237 :
2238 1 : void* raReqHandle = nullptr;
2239 1 : int ret = RaSocketBatchCloseAsync(&closeInfo, SOCKET_NUM_ONE, &raReqHandle);
2240 1 : if (ret != 0) {
2241 0 : MACRO_THROW(
2242 : NetworkApiException, StringFormat(
2243 : "[BatchClose][RaSocket]errNo[0x%016llx] ra socket batch close fail. return[%d]",
2244 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret));
2245 : }
2246 :
2247 1 : return reinterpret_cast<RequestHandle>(raReqHandle);
2248 : }
2249 :
2250 3 : RequestHandle RaSocketListenOneStartAsync(SocketListenInfoT* listenInfo)
2251 : {
2252 3 : if (listenInfo == nullptr) {
2253 0 : MACRO_THROW(
2254 : NetworkApiException,
2255 : StringFormat("errNo[0x%016llx] listenInfo is nullptr.", HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT)));
2256 : }
2257 9 : HCCL_INFO("[RaSocketListenOneStartAsync] Input params: listenInfo=%p, port=%u", listenInfo, listenInfo->port);
2258 :
2259 3 : void* raReqHandle = nullptr;
2260 3 : int ret = RaSocketListenStartAsync(listenInfo, SOCKET_NUM_ONE, &raReqHandle);
2261 3 : if (ret != 0) {
2262 0 : MACRO_THROW(
2263 : NetworkApiException, StringFormat(
2264 : "errNo[0x%016llx] ra socket listen start fail. return[%d]",
2265 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret));
2266 : }
2267 :
2268 3 : return reinterpret_cast<RequestHandle>(raReqHandle);
2269 : }
2270 :
2271 1 : RequestHandle RaSocketListenOneStopAsync(RaSocketListenParam& in)
2272 : {
2273 3 : HCCL_INFO("[RaSocketListenOneStopAsync] Input params: socketHandle=%p, port=%u", in.socketHandle, in.port);
2274 1 : struct SocketListenInfoT listenInfo {};
2275 1 : listenInfo.socketHandle = in.socketHandle;
2276 1 : listenInfo.port = in.port;
2277 :
2278 1 : void* raReqHandle = nullptr;
2279 1 : int ret = RaSocketListenStopAsync(&listenInfo, SOCKET_NUM_ONE, &raReqHandle);
2280 1 : if (ret != 0) {
2281 0 : MACRO_THROW(
2282 : NetworkApiException, StringFormat(
2283 : "[ListenStop][RaSocket]errNo[0x%016llx] ra socket listen stop fail. return[%d]",
2284 : HCCL_ERROR_CODE(HcclResult::HCCL_E_TCP_CONNECT), ret));
2285 : }
2286 :
2287 1 : return reinterpret_cast<RequestHandle>(raReqHandle);
2288 : }
2289 :
2290 11 : RaSocketFdHandleParam RaGetOneSocket(u32 role, RaSocketGetParam& param)
2291 : {
2292 11 : struct SocketInfoT socketInfo {};
2293 :
2294 11 : socketInfo.socketHandle = param.socketHandle;
2295 11 : socketInfo.fdHandle = param.fdHandle;
2296 11 : socketInfo.remoteIp = IpAddressToHccpIpAddr(param.remoteIp);
2297 11 : socketInfo.status = SOCKET_NOT_CONNECTED;
2298 :
2299 11 : int sret = strcpy_s(socketInfo.tag, sizeof(socketInfo.tag), param.tag.c_str());
2300 11 : if (sret != 0) {
2301 4 : MACRO_THROW(
2302 : NetworkApiException,
2303 : StringFormat(
2304 : "[%s] failed, copy tag[%s] to hccp failed, ret=%d, socketInfo.tag size=%zu, param.tag size=%zu",
2305 : __func__, param.tag.c_str(), sret, sizeof(socketInfo.tag), sizeof(param.tag.c_str())));
2306 : }
2307 :
2308 10 : u32 connectedNum = 0;
2309 10 : s32 sockRet = RaGetSockets(role, &socketInfo, SOCKET_NUM_ONE, &connectedNum);
2310 10 : if ((connectedNum == 0 && sockRet == 0) || sockRet == SOCK_EAGAIN) {
2311 : // 更新为 connecting 状态，表示连接未完成
2312 0 : socketInfo.status = SOCKET_CONNECTING;
2313 0 : return RaSocketFdHandleParam(socketInfo.fdHandle, socketInfo.status);
2314 : }
2315 :
2316 10 : if (sockRet != 0) {
2317 0 : MACRO_THROW(
2318 : NetworkApiException, StringFormat(
2319 : "[%s] failed, call interface error[%d], "
2320 : "role[%u], num[%u], connectednum[%u]",
2321 : __func__, sockRet, role, SOCKET_NUM_ONE, connectedNum));
2322 : }
2323 :
2324 10 : if (connectedNum > SOCKET_NUM_ONE) {
2325 4 : MACRO_THROW(
2326 : NetworkApiException,
2327 : StringFormat(
2328 : "[%s] failed, connetedNum[%u] is more "
2329 : "than expected[%u], role[%u], num[%u], connectednum[%u]",
2330 : __func__, connectedNum, SOCKET_NUM_ONE, sockRet, role, SOCKET_NUM_ONE, connectedNum));
2331 : }
2332 :
2333 9 : return RaSocketFdHandleParam(socketInfo.fdHandle, socketInfo.status);
2334 : }
2335 :
2336 5 : RequestHandle HrtRaSocketSendAsync(const FdHandle fdHandle, const void* data, u32 size, unsigned long long& sentSize)
2337 : {
2338 10 : CHECK_NULLPTR(fdHandle, "[HrtRaSocketSendAsync] fdHandle is nullptr!");
2339 5 : CHECK_NULLPTR(data, "[HrtRaSocketSendAsync] data is nullptr!");
2340 15 : HCCL_INFO(
2341 : "[HrtRaSocketSendAsync] Input params: fdHandle=%p, data=%p, size=%u, sentSize=%llu", fdHandle, data, size,
2342 : sentSize);
2343 5 : void* raReqHandle = nullptr;
2344 5 : s32 ret = RaSocketSendAsync(fdHandle, data, size, &sentSize, &raReqHandle);
2345 5 : if (ret != 0 || !raReqHandle) {
2346 0 : MACRO_THROW(
2347 : NetworkApiException, StringFormat(
2348 : "[%s] failed, call interface error[%d] "
2349 : "raReqHandle[%p], fdHandle[%p], data[%p], size[%u], sentSize[%u].",
2350 : __func__, ret, raReqHandle, fdHandle, data, size, sentSize));
2351 : }
2352 :
2353 5 : return reinterpret_cast<RequestHandle>(raReqHandle);
2354 : }
2355 :
2356 5 : RequestHandle HrtRaSocketRecvAsync(const FdHandle fdHandle, void* data, u32 size, unsigned long long& recvSize)
2357 : {
2358 10 : CHECK_NULLPTR(fdHandle, "[HrtRaSocketRecvAsync] fdHandle is nullptr!");
2359 5 : CHECK_NULLPTR(data, "[HrtRaSocketRecvAsync] data is nullptr!");
2360 15 : HCCL_INFO(
2361 : "[HrtRaSocketRecvAsync] Input params: fdHandle=%p, data=%p, size=%u, recvSize=%llu", fdHandle, data, size,
2362 : recvSize);
2363 5 : void* raReqHandle = nullptr;
2364 5 : s32 ret = RaSocketRecvAsync(fdHandle, data, size, &recvSize, &raReqHandle);
2365 5 : if (ret != 0 || !raReqHandle) {
2366 0 : MACRO_THROW(
2367 : NetworkApiException, StringFormat(
2368 : "[%s] failed, call interface error[%d], "
2369 : "raReqHandle[%p], fdHandle[%p], data[%p], size[%u], recvSize[%u].",
2370 : __func__, ret, raReqHandle, fdHandle, data, size, recvSize));
2371 : }
2372 :
2373 5 : return reinterpret_cast<RequestHandle>(raReqHandle);
2374 : }
2375 :
2376 : RequestHandle
2377 1 : RaUbLocalMemRegAsync(RdmaHandle handle, const HrtRaUbLocMemRegParam& in, vector<char_t>& out, void*& lmemHandle)
2378 : {
2379 2 : CHECK_NULLPTR(handle, "[RaUbLocalMemRegAsync] handle is nullptr!");
2380 1 : CHECK_NULLPTR(lmemHandle, "[RaUbLocalMemRegAsync] lmemHandle is nullptr!");
2381 3 : HCCL_INFO(
2382 : "[RaUbLocalMemRegAsync] Input params: handle=%p, addr=0x%llx, size=0x%llx, lmemHandle=%p", handle, in.addr,
2383 : in.size, lmemHandle);
2384 1 : u64 pageSize = UB_MEM_PAGE_SIZE;
2385 1 : u64 newAddr = in.addr & (~(static_cast<u64>(pageSize - 1))); // UB内存注册要求起始地址4k对齐
2386 1 : u64 offset = in.addr - newAddr;
2387 1 : u64 newSize = in.size + offset + 4;
2388 :
2389 1 : out.resize(sizeof(struct MrRegInfoT));
2390 1 : struct MrRegInfoT* info = reinterpret_cast<struct MrRegInfoT*>(out.data());
2391 1 : info->in.mem.addr = newAddr;
2392 1 : info->in.mem.size = newSize;
2393 :
2394 1 : info->in.ub.flags.value = 0;
2395 1 : info->in.ub.flags.bs.tokenPolicy = TOKEN_POLICY_PLAIN_TEXT;
2396 1 : info->in.ub.flags.bs.tokenIdValid = 1;
2397 1 : info->in.ub.flags.bs.access = MEM_SEG_ACCESS_READ | MEM_SEG_ACCESS_WRITE | MEM_SEG_ACCESS_ATOMIC;
2398 1 : info->in.ub.flags.bs.nonPin = in.nonPin;
2399 1 : info->in.ub.tokenValue = in.tokenValue;
2400 1 : info->in.ub.tokenIdHandle = reinterpret_cast<void*>(in.tokenIdHandle);
2401 :
2402 1 : void* raReqHandle = nullptr;
2403 1 : s32 ret = RaCtxLmemRegisterAsync(handle, info, &lmemHandle, &raReqHandle);
2404 1 : if (ret != 0 || !raReqHandle) {
2405 0 : MACRO_THROW(
2406 : NetworkApiException, StringFormat(
2407 : "[%s] failed, call interface "
2408 : "error[%d], raReqHandle[%p], addr=0x%llx, size=0x%llx",
2409 : __func__, ret, raReqHandle, in.addr, in.size));
2410 : }
2411 1 : info->in.ub.tokenValue = 0;
2412 3 : HCCL_INFO(
2413 : "[%s] RaCtxLmemRegisterAsync success, reqHandle[%llu] addr[0x%llx] size[0x%llx].", __func__,
2414 : reinterpret_cast<RequestHandle>(raReqHandle), in.addr, in.size);
2415 1 : return reinterpret_cast<RequestHandle>(raReqHandle);
2416 : }
2417 :
2418 1 : RequestHandle RaUbLocalMemUnregAsync(RdmaHandle rdmaHandle, LocMemHandle lmemHandle)
2419 : {
2420 1 : CHECK_NULLPTR(rdmaHandle, "[RaUbLocalMemUnregAsync] rdmaHandle is nullptr!");
2421 3 : HCCL_INFO("[RaUbLocalMemUnregAsync] Input params: rdmaHandle=%p, lmemHandle=0x%llx", rdmaHandle, lmemHandle);
2422 1 : void* raReqHandle = nullptr;
2423 1 : s32 ret = RaCtxLmemUnregisterAsync(rdmaHandle, reinterpret_cast<void*>(lmemHandle), &raReqHandle);
2424 1 : if (ret != 0 || !raReqHandle) {
2425 0 : MACRO_THROW(
2426 : NetworkApiException, StringFormat(
2427 : "[%s] failed, call interface error[%d] "
2428 : "raReqResult[%p], rdmaHandle=%p, lmemHandle=0x%llx.",
2429 : __func__, ret, raReqHandle, rdmaHandle, lmemHandle));
2430 : }
2431 :
2432 3 : HCCL_INFO(
2433 : "[%s] RaCtxLmemUnregisterAsync success, reqHandle[%llu] lmemHandle[0x%llx].", __func__,
2434 : reinterpret_cast<RequestHandle>(raReqHandle), lmemHandle);
2435 1 : return reinterpret_cast<RequestHandle>(raReqHandle);
2436 : }
2437 :
2438 110 : RequestHandle RaUbCreateJettyAsync(
2439 : const RdmaHandle handle, const HrtRaUbCreateJettyParam& in, vector<char_t>& out, void*& jettyHandle)
2440 : {
2441 110 : struct QpCreateAttr attr = GetQpCreateAttr(in);
2442 :
2443 110 : void* raReqHandle = nullptr;
2444 110 : out.resize(sizeof(QpCreateInfo));
2445 : s32 ret
2446 110 : = RaCtxQpCreateAsync(handle, &attr, reinterpret_cast<QpCreateInfo*>(out.data()), &jettyHandle, &raReqHandle);
2447 110 : if (ret != 0 || !raReqHandle) {
2448 0 : MACRO_THROW(
2449 : NetworkApiException, StringFormat(
2450 : "[%s] failed, call interface error[%d], raReqHandle[%p], "
2451 : "rdmaHanlde[%p].",
2452 : __func__, ret, raReqHandle, handle));
2453 : }
2454 110 : attr.ub.tokenValue = 0;
2455 330 : HCCL_INFO(
2456 : "[%s] RaCtxQpCreateAsync success, reqHandle[%llu] jettyHandle[%p].", __func__,
2457 : reinterpret_cast<RequestHandle>(raReqHandle), jettyHandle);
2458 110 : return reinterpret_cast<RequestHandle>(raReqHandle);
2459 : }
2460 :
2461 1 : RequestHandle RaUbDestroyJettyAsync(void* jettyHandle)
2462 : {
2463 1 : CHECK_NULLPTR(jettyHandle, "[RaUbDestroyJettyAsync] jettyHandle is nullptr!");
2464 3 : HCCL_INFO("[RaUbDestroyJettyAsync] Input params: jettyHandle=%p", jettyHandle);
2465 1 : void* raReqHandle = nullptr;
2466 1 : s32 ret = RaCtxQpDestroyAsync(jettyHandle, &raReqHandle);
2467 1 : if (ret != 0) {
2468 0 : MACRO_THROW(
2469 : NetworkApiException, StringFormat(
2470 : "[%s] failed, call interface error[%d] raReqHandle[%p], "
2471 : "jettyHandle[%p].",
2472 : __func__, ret, raReqHandle, jettyHandle));
2473 : }
2474 :
2475 3 : HCCL_INFO(
2476 : "[%s] RaCtxQpDestroyAsync success, reqHandle[%llu] jettyHandle[%p].", __func__,
2477 : reinterpret_cast<RequestHandle>(raReqHandle), jettyHandle);
2478 1 : return reinterpret_cast<RequestHandle>(raReqHandle);
2479 : }
2480 :
2481 32 : inline HccpEid IpAddressToHccpEid(const IpAddress& ipAddr)
2482 : {
2483 32 : HccpEid eid = {};
2484 96 : HCCL_INFO("EID ipAddr[%s]", ipAddr.Describe().c_str());
2485 32 : s32 sRet = memcpy_s(eid.raw, sizeof(eid.raw), ipAddr.GetEid().raw, sizeof(ipAddr.GetEid().raw));
2486 32 : if (sRet != EOK) {
2487 0 : MACRO_THROW(
2488 : InternalException,
2489 : StringFormat(
2490 : "[IpAddressToHccpEid]memcpy_s failed. sRet[%d], dest[%p], destSize[%zu], src[%p], srcSize[%zu]", sRet,
2491 : eid.raw, sizeof(eid.raw), ipAddr.GetEid().raw, sizeof(ipAddr.GetEid().raw)));
2492 : }
2493 96 : HCCL_INFO("[IpAddressToHccpEid] %s", HccpEidDesc(eid).c_str());
2494 32 : return eid;
2495 : }
2496 :
2497 : RequestHandle
2498 16 : RaUbGetTpInfoAsync(const RdmaHandle rdmaHandle, const RaUbGetTpInfoParam& param, vector<char_t>& out, uint32_t& num)
2499 : {
2500 16 : CHECK_NULLPTR(rdmaHandle, "[RaUbGetTpInfoAsync] rdmaHandle is nullptr!");
2501 48 : HCCL_INFO("[RaUbGetTpInfoAsync] Input params: rdmaHandle=%p, num=%u", rdmaHandle, num);
2502 16 : const auto& locAddr = param.locAddr;
2503 16 : const auto& rmtAddr = param.rmtAddr;
2504 16 : const auto& tpProtocol = param.tpProtocol;
2505 :
2506 16 : struct GetTpCfg cfg {};
2507 : // UBG与TP同属RTP传输，需使能rtp位；UBOE走独立uboe位
2508 16 : cfg.flag.bs.rtp = (tpProtocol == TpProtocol::TP || tpProtocol == TpProtocol::UBG) ? 1 : 0;
2509 16 : cfg.flag.bs.ctp = tpProtocol == TpProtocol::CTP ? 1 : 0;
2510 16 : cfg.flag.bs.uboe = (tpProtocol == TpProtocol::UBOE) ? 1 : 0;
2511 16 : cfg.transMode = TransportModeT::CONN_RM; // 当前只使用RM Jetty
2512 16 : cfg.localEid = IpAddressToHccpEid(locAddr);
2513 48 : HCCL_INFO("RaUbGetTpInfoAsync cfg.localEid=%s", HccpEidDesc(cfg.localEid).c_str());
2514 16 : cfg.peerEid = IpAddressToHccpEid(rmtAddr);
2515 48 : HCCL_INFO("RaUbGetTpInfoAsync cfg.peerEid=%s", HccpEidDesc(cfg.peerEid).c_str());
2516 :
2517 : // 须至少容纳 TP_HANDLE_REQUEST_NUM 条 HccpTpInfo，避免 RS 按 num 写多条时越界破坏堆
2518 16 : out.resize(static_cast<size_t>(TP_HANDLE_REQUEST_NUM) * sizeof(struct HccpTpInfo));
2519 16 : struct HccpTpInfo* info = reinterpret_cast<struct HccpTpInfo*>(out.data());
2520 :
2521 16 : void* raReqHandle = nullptr;
2522 16 : num = TP_HANDLE_REQUEST_NUM; // 指定需要从管控面申请tp handle的数量, hccp 会返回实际个数
2523 16 : s32 ret = RaGetTpInfoListAsync(rdmaHandle, &cfg, info, &num, &raReqHandle);
2524 16 : if (ret != 0 || !raReqHandle) {
2525 4 : MACRO_THROW(
2526 : NetworkApiException,
2527 : StringFormat(
2528 : "[%s] failed, call interface error[%d] raReqHandle[%p], "
2529 : "rdmaHandle[%p], locAddr[%s], rmtAddr[%s].",
2530 : __func__, ret, raReqHandle, rdmaHandle, locAddr.Describe().c_str(), rmtAddr.Describe().c_str()));
2531 : }
2532 :
2533 45 : HCCL_INFO(
2534 : "[%s] RaGetTpInfoListAsync success, reqHandle[%llu] locAddr[%s] rmtAddr[%s] tpNum[%u].", __func__,
2535 : reinterpret_cast<RequestHandle>(raReqHandle), locAddr.Describe().c_str(), rmtAddr.Describe().c_str(), num);
2536 15 : return reinterpret_cast<RequestHandle>(raReqHandle);
2537 : }
2538 :
2539 0 : void RaUbGetTpInfo(const RdmaHandle rdmaHandle, const RaUbGetTpInfoParam& param, vector<char_t>& out, uint32_t& num)
2540 : {
2541 0 : CHECK_NULLPTR(rdmaHandle, "[RaUbGetTpInfo] rdmaHandle is nullptr!");
2542 0 : HCCL_INFO("[RaUbGetTpInfo] Input params: rdmaHandle=%p, num=%u", rdmaHandle, num);
2543 0 : const auto& locAddr = param.locAddr;
2544 0 : const auto& rmtAddr = param.rmtAddr;
2545 0 : const auto& tpProtocol = param.tpProtocol;
2546 :
2547 0 : struct GetTpCfg cfg {};
2548 : // UBG与TP同属RTP传输，需使能rtp位；UBOE走独立uboe位
2549 0 : cfg.flag.bs.rtp = (tpProtocol == TpProtocol::TP || tpProtocol == TpProtocol::UBG) ? 1 : 0;
2550 0 : cfg.flag.bs.ctp = tpProtocol == TpProtocol::CTP ? 1 : 0;
2551 0 : cfg.transMode = TransportModeT::CONN_RM; // 当前只使用RM Jetty
2552 0 : cfg.localEid = IpAddressToHccpEid(locAddr);
2553 0 : HCCL_INFO("RaUbGetTpInfo cfg.localEid=%s", HccpEidDesc(cfg.localEid).c_str());
2554 0 : cfg.peerEid = IpAddressToHccpEid(rmtAddr);
2555 0 : HCCL_INFO("RaUbGetTpInfo cfg.peerEid=%s", HccpEidDesc(cfg.peerEid).c_str());
2556 :
2557 0 : out.resize(static_cast<size_t>(TP_HANDLE_REQUEST_NUM) * sizeof(struct HccpTpInfo));
2558 0 : struct HccpTpInfo* info = reinterpret_cast<struct HccpTpInfo*>(out.data());
2559 :
2560 0 : num = TP_HANDLE_REQUEST_NUM; // 指定需要从管控面申请tp handle的数量, hccp 会返回实际个数
2561 0 : s32 ret = RaCtxGetTpInfoList(rdmaHandle, &cfg, info, &num);
2562 0 : if (ret != 0) {
2563 0 : MACRO_THROW(
2564 : NetworkApiException,
2565 : StringFormat(
2566 : "[%s] failed, call interface error[%d], "
2567 : "rdmaHandle[%p], locAddr[%s], rmtAddr[%s].",
2568 : __func__, ret, rdmaHandle, locAddr.Describe().c_str(), rmtAddr.Describe().c_str()));
2569 : }
2570 :
2571 0 : HCCL_INFO(
2572 : "[%s] RaCtxGetTpInfoList success, locAddr[%s] rmtAddr[%s] tpNum[%u].", __func__, locAddr.Describe().c_str(),
2573 : rmtAddr.Describe().c_str(), num);
2574 0 : }
2575 :
2576 17 : static RequestHandle ImportJettyAsync(
2577 : RdmaHandle rdmaHandle, const HrtRaUbJettyImportedInParam& in, vector<char_t>& out, void*& remQpHandle,
2578 : const JettyImportExpCfg& cfg, JettyImportMode mode, TpProtocol protocol = TpProtocol::INVALID)
2579 : {
2580 17 : CHECK_NULLPTR(rdmaHandle, "[ImportJettyAsync] rdmaHandle is nullptr!");
2581 51 : HCCL_INFO("[ImportJettyAsync] Input params: rdmaHandle=%p, remQpHandle=%p", rdmaHandle, remQpHandle);
2582 17 : if (mode == JettyImportMode::JETTY_IMPORT_MODE_NORMAL) {
2583 0 : MACRO_THROW(
2584 : NotSupportException, StringFormat("[%s] currently not support JETTY_IMPORT_MODE_NORMAL.", __func__));
2585 : }
2586 :
2587 17 : out.resize(sizeof(QpImportInfoT));
2588 17 : struct QpImportInfoT* info = reinterpret_cast<QpImportInfoT*>(out.data());
2589 :
2590 17 : s32 ret = memcpy_s(info->in.key.value, sizeof(info->in.key.value), in.key, in.keyLen);
2591 17 : if (ret != 0) {
2592 0 : MACRO_THROW(InternalException, StringFormat("[%s] memcpy_s failed, ret=%d.", __func__, ret));
2593 : }
2594 :
2595 17 : info->in.key.size = in.keyLen;
2596 17 : info->in.ub.mode = mode;
2597 17 : info->in.ub.tokenValue = in.tokenValue;
2598 17 : info->in.ub.policy = JettyGrpPolicy::JETTY_GRP_POLICY_RR;
2599 17 : info->in.ub.type = TargetType::TARGET_TYPE_JETTY;
2600 :
2601 17 : info->in.ub.flag.value = 0;
2602 17 : info->in.ub.flag.bs.tokenPolicy = TOKEN_POLICY_PLAIN_TEXT;
2603 :
2604 17 : info->in.ub.expImportCfg = cfg;
2605 :
2606 31 : if (protocol != TpProtocol::TP && protocol != TpProtocol::CTP && protocol != TpProtocol::UBOE
2607 31 : && protocol != TpProtocol::UBG) {
2608 0 : MACRO_THROW(
2609 : NetworkApiException,
2610 : StringFormat("[%s] failed, tp protocol[%s] is not expected, %s.", __func__, protocol.Describe().c_str()));
2611 : }
2612 : // tpType: 0->RTP, 1->CTP
2613 17 : info->in.ub.tpType = protocol == TpProtocol::TP ? 0 : 1;
2614 :
2615 17 : void* raReqHandle = nullptr;
2616 17 : ret = RaCtxQpImportAsync(rdmaHandle, info, &remQpHandle, &raReqHandle);
2617 17 : if (ret != 0 || !raReqHandle) {
2618 0 : MACRO_THROW(
2619 : NetworkApiException, StringFormat(
2620 : "[%s] failed, call interface error[%d] raReqHandle[%p], "
2621 : "rdmaHandle[%p].",
2622 : __func__, ret, raReqHandle, rdmaHandle));
2623 : }
2624 17 : info->in.ub.tokenValue = 0;
2625 51 : HCCL_INFO(
2626 : "[%s] RaCtxQpImportAsync success, reqHandle[%llu] remQpHandle[%p].", __func__,
2627 : reinterpret_cast<RequestHandle>(raReqHandle), remQpHandle);
2628 17 : return reinterpret_cast<RequestHandle>(raReqHandle);
2629 : }
2630 :
2631 0 : RequestHandle RaUbImportJettyAsync(
2632 : const RdmaHandle rdmaHandle, const HrtRaUbJettyImportedInParam& in, vector<char_t>& out, void*& remQpHandle)
2633 : {
2634 0 : CHECK_NULLPTR(rdmaHandle, "[RaUbImportJettyAsync] rdmaHandle is nullptr!");
2635 0 : HCCL_INFO("[RaUbImportJettyAsync] Input params: rdmaHandle=%p, remQpHandle=%p", rdmaHandle, remQpHandle);
2636 : // 该接口仅适配非管控面模式，当前不期望使用
2637 0 : struct JettyImportExpCfg cfg = {};
2638 0 : const auto mode = JettyImportMode::JETTY_IMPORT_MODE_NORMAL;
2639 0 : return ImportJettyAsync(rdmaHandle, in, out, remQpHandle, cfg, mode);
2640 : }
2641 :
2642 17 : RequestHandle RaUbTpImportJettyAsync(
2643 : const RdmaHandle rdmaHandle, const HrtRaUbJettyImportedInParam& in, vector<char_t>& out, void*& remQpHandle)
2644 : {
2645 17 : CHECK_NULLPTR(rdmaHandle, "[RaUbTpImportJettyAsync] rdmaHandle is nullptr!");
2646 51 : HCCL_INFO("[RaUbTpImportJettyAsync] Input params: rdmaHandle=%p, remQpHandle=%p", rdmaHandle, remQpHandle);
2647 17 : struct JettyImportExpCfg cfg = GetTpImportCfg(in.jettyImportCfg);
2648 17 : const auto mode = JettyImportMode::JETTY_IMPORT_MODE_EXP;
2649 34 : return ImportJettyAsync(rdmaHandle, in, out, remQpHandle, cfg, mode, in.jettyImportCfg.protocol);
2650 : }
2651 :
2652 1 : RequestHandle RaUbUnimportJettyAsync(void* targetJettyHandle)
2653 : {
2654 1 : CHECK_NULLPTR(targetJettyHandle, "[RaUbUnimportJettyAsync] targetJettyHandle is nullptr!");
2655 3 : HCCL_INFO("[RaUbUnimportJettyAsync] Input params: targetJettyHandle=%p", targetJettyHandle);
2656 1 : void* raReqHandle = nullptr;
2657 1 : s32 ret = RaCtxQpUnimportAsync(targetJettyHandle, &raReqHandle);
2658 1 : if (ret != 0 || !raReqHandle) {
2659 0 : MACRO_THROW(
2660 : NetworkApiException, StringFormat(
2661 : "[%s] failed, call interface error[%d] raReqHandle[%p], "
2662 : "targetJettyHandle[%p].",
2663 : __func__, ret, raReqHandle, targetJettyHandle));
2664 : }
2665 :
2666 3 : HCCL_INFO(
2667 : "[%s] RaCtxQpUnimportAsync success, reqHandle[%llu] targetJettyHandle[%p].", __func__,
2668 : reinterpret_cast<RequestHandle>(raReqHandle), targetJettyHandle);
2669 1 : return reinterpret_cast<RequestHandle>(raReqHandle);
2670 : }
2671 :
2672 1 : HcclResult HrtRaWaitEventHandle(
2673 : int event_handle, std::vector<SocketEventInfo>& event_infos, int timeout, unsigned int maxevents, u32& events_num)
2674 : {
2675 3 : HCCL_INFO(
2676 : "[HrtRaWaitEventHandle] Input params: event_handle=[%d], timeout=[%d ms], maxevents=[%u], events_num=[%u]",
2677 : event_handle, timeout, maxevents, events_num);
2678 1 : std::vector<struct SocketEventInfoT> raEventInfos(maxevents);
2679 1 : s32 ret = RaWaitEventHandle(event_handle, raEventInfos.data(), timeout, maxevents, &events_num);
2680 1 : CHK_PRT_RET(
2681 : ret != 0, HCCL_ERROR("[%s] failed, call RaWaitEventHandle error ret[%d].", __func__, ret), HCCL_E_NETWORK);
2682 1 : for (u32 i = 0; i < events_num; i++) {
2683 0 : event_infos[i].fdHandle = raEventInfos[i].fdHandle;
2684 : }
2685 1 : return HCCL_SUCCESS;
2686 1 : }
2687 :
2688 1 : void HrtRaGetSecRandom(u32* value, u32& devPhyId)
2689 : {
2690 1 : CHECK_NULLPTR(value, "[HrtRaGetSecRandom] value is nullptr!");
2691 3 : HCCL_INFO("[HrtRaGetSecRandom] Input params: value=%u, devPhyId=%u", *value, devPhyId);
2692 1 : struct RaInfo raInfo = {};
2693 1 : raInfo.mode = HrtNetworkMode::HDC;
2694 1 : raInfo.phyId = devPhyId;
2695 :
2696 1 : s32 ret = RaGetSecRandom(&raInfo, value);
2697 1 : if (ret != 0) {
2698 0 : MACRO_THROW(
2699 : NetworkApiException, StringFormat(
2700 : "[%s] failed, call interface error[%d]. params: value=%u, devPhyId=%u", __func__,
2701 : ret, *value, devPhyId));
2702 : }
2703 3 : HCCL_INFO("[HrtRaGetSecRandom] Input params: value=%u, devPhyId=%u", *value, devPhyId);
2704 1 : }
2705 0 : HcclResult HrtRaCreateQpWithCq(
2706 : RdmaHandle rdmaHandle, s32 sqEvent, s32 rqEvent, void* sendChannel, void* recvChannel, QpInfo& info, bool isHdcMode)
2707 : {
2708 0 : CHK_PTR_NULL(rdmaHandle);
2709 0 : CHK_PTR_NULL(sendChannel);
2710 0 : CHK_PTR_NULL(recvChannel);
2711 0 : HCCL_INFO(
2712 : "[HrtRaCreateQpWithCq] Input params: rdmaHandle=%p, sqEvent=%d, rqEvent=%d, sendChannel=%p, recvChannel=%p",
2713 : rdmaHandle, sqEvent, rqEvent, sendChannel, recvChannel);
2714 0 : struct ibv_comp_channel* sChannel = reinterpret_cast<struct ibv_comp_channel*>(sendChannel);
2715 0 : struct ibv_comp_channel* rChannel = reinterpret_cast<struct ibv_comp_channel*>(recvChannel);
2716 :
2717 0 : QpConfig config(MAX_WR_NUM, MAX_SEND_SGE_NUM, MAX_RECV_SGE_NUM, sqEvent, rqEvent);
2718 : CqInfo cq(
2719 0 : nullptr, nullptr, nullptr, MAX_CQ_DEPTH, config.sqEvent, config.rqEvent, info.srqContext, sChannel, rChannel);
2720 : // hdc模式下hccp没有对外提供创建CQ的接口
2721 0 : if (!isHdcMode) {
2722 0 : CHK_RET(HrtRaCreateCq(rdmaHandle, cq));
2723 : }
2724 0 : info.attr = config;
2725 0 : info.rdmaHandle = rdmaHandle;
2726 0 : info.context = cq.context;
2727 0 : info.sendCq = cq.sq;
2728 0 : info.recvCq = cq.rq;
2729 0 : info.recvChannel = rChannel;
2730 0 : info.sendChannel = sChannel;
2731 :
2732 0 : if (isHdcMode) {
2733 0 : TRY_CATCH_RETURN(info.qpHandle = HrtRaQpCreate(rdmaHandle, info.flag, info.qpMode));
2734 : } else {
2735 0 : CHK_RET(HrtRaNormalQpCreate(rdmaHandle, info));
2736 : }
2737 :
2738 0 : return HCCL_SUCCESS;
2739 0 : }
2740 :
2741 0 : HcclResult HrtRaDestroyQpWithCq(const QpInfo& info, bool isHdcMode)
2742 : {
2743 0 : if (info.qpHandle == nullptr) {
2744 0 : return HCCL_SUCCESS;
2745 : }
2746 :
2747 0 : if (isHdcMode) {
2748 0 : TRY_CATCH_RETURN(HrtRaQpDestroy(info.qpHandle));
2749 : } else {
2750 0 : CHK_RET(HrtRaNormalQpDestroy(info.qpHandle));
2751 0 : CqInfo cq;
2752 0 : cq.context = info.context;
2753 0 : cq.rq = info.recvCq;
2754 0 : cq.sq = info.sendCq;
2755 0 : CHK_RET(HrtRaDestroyCq(info.rdmaHandle, cq));
2756 0 : }
2757 :
2758 0 : return HCCL_SUCCESS;
2759 : }
2760 :
2761 : // ra_cq_create
2762 0 : HcclResult HrtRaCreateCq(RdmaHandle rdmaHandle, CqInfo& cq)
2763 : {
2764 0 : CHK_PTR_NULL(rdmaHandle);
2765 0 : HCCL_INFO(
2766 : "[HrtRaCreateCq] Input params: rdmaHandle=%p, sq=%p, rq=%p, context=%p", rdmaHandle, cq.sq, cq.rq, cq.context);
2767 :
2768 0 : struct CqAttr attr {};
2769 0 : attr.qpContext = &(cq.context);
2770 0 : attr.ibSendCq = &(cq.sq);
2771 0 : attr.ibRecvCq = &(cq.rq);
2772 0 : attr.sendCqDepth = cq.depth;
2773 0 : attr.recvCqDepth = cq.depth;
2774 0 : attr.sendCqEventId = cq.sqEvent;
2775 0 : attr.recvCqEventId = cq.rqEvent;
2776 0 : attr.sendChannel = cq.sendChannel;
2777 0 : attr.recvChannel = cq.recvChannel;
2778 0 : attr.srqContext = cq.srqContext;
2779 :
2780 0 : HCCL_DEBUG(
2781 : "ra create cq: send_cq_depth[%d], recv_cq_depth[%d], send_cq_event_id[%d], recv_cq_event_id[%d]",
2782 : attr.sendCqDepth, attr.recvCqDepth, attr.sendCqEventId, attr.recvCqEventId);
2783 0 : s32 ret = RaCqCreate(rdmaHandle, &attr);
2784 0 : CHK_PRT_RET(
2785 : ret != 0,
2786 : HCCL_ERROR(
2787 : "[HrtRaCreateCq] errNo[0x%016llx] RaCqCreate fail. "
2788 : "return[%d], params: rdmaHandle[%p], sq[%p], rq[%p], context[%p]",
2789 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, rdmaHandle, cq.sq, cq.rq, cq.context),
2790 : HCCL_E_NETWORK);
2791 0 : if (cq.sq == nullptr || cq.rq == nullptr || cq.context == nullptr) {
2792 0 : HCCL_ERROR(
2793 : "[HrtRaCreateCq] cq member[sq:%p, rq:%p, context:%p] is nullptr, ret[%d]", cq.sq, cq.rq, cq.context, ret);
2794 0 : return HCCL_E_PARA;
2795 : }
2796 0 : return HCCL_SUCCESS;
2797 : }
2798 : // ra_cq_destroy
2799 0 : HcclResult HrtRaDestroyCq(RdmaHandle rdmaHandle, CqInfo& cq)
2800 : {
2801 0 : CHK_PTR_NULL(rdmaHandle);
2802 0 : HCCL_INFO(
2803 : "[HrtRaDestroyCq] Input params: rdmaHandle=%p, sq=%p, rq=%p, context=%p", rdmaHandle, cq.sq, cq.rq, cq.context);
2804 0 : struct CqAttr attr = {};
2805 0 : attr.qpContext = &cq.context;
2806 0 : attr.ibSendCq = &cq.sq;
2807 0 : attr.ibRecvCq = &cq.rq;
2808 0 : s32 ret = RaCqDestroy(rdmaHandle, &attr);
2809 0 : CHK_PRT_RET(
2810 : ret != 0,
2811 : HCCL_ERROR(
2812 : "[HrtRaDestroyCq] errNo[0x%016llx] RaCqDestroy failed, call interface error. "
2813 : "return[%d], params: rdmaHandle[%p], sq[%p], rq[%p], context[%p]",
2814 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, rdmaHandle, cq.sq, cq.rq, cq.context),
2815 : HCCL_E_NETWORK);
2816 0 : return HCCL_SUCCESS;
2817 : }
2818 :
2819 : // ra_normal_qp_create
2820 0 : HcclResult HrtRaNormalQpCreate(RdmaHandle rdmaHandle, QpInfo& qp)
2821 : {
2822 0 : CHK_PTR_NULL(rdmaHandle);
2823 0 : HCCL_INFO("[HrtRaNormalQpCreate] Input params: rdmaHandle=%p, context=%p", rdmaHandle, qp.context);
2824 0 : struct ibv_qp_init_attr ibQpAttr = {};
2825 0 : CHK_SAFETY_FUNC_RET(memset_s(&ibQpAttr, sizeof(ibv_qp_init_attr), 0, sizeof(ibv_qp_init_attr)));
2826 0 : ibQpAttr.qp_context = qp.context;
2827 0 : ibQpAttr.send_cq = qp.sendCq;
2828 0 : ibQpAttr.recv_cq = qp.recvCq;
2829 0 : ibQpAttr.srq = qp.srq;
2830 0 : ibQpAttr.qp_type = IBV_QPT_RC;
2831 0 : ibQpAttr.cap.max_inline_data = MAX_INLINE_DATA;
2832 0 : ibQpAttr.cap.max_send_wr = qp.attr.maxWr;
2833 0 : ibQpAttr.cap.max_send_sge = qp.attr.maxSendSge;
2834 0 : ibQpAttr.cap.max_recv_wr = (qp.srq == nullptr ? qp.attr.maxWr : 0);
2835 0 : ibQpAttr.cap.max_recv_sge = (qp.srq == nullptr ? qp.attr.maxRecvSge : 0);
2836 0 : s32 ret = RaNormalQpCreate(rdmaHandle, &ibQpAttr, &(qp.qpHandle), reinterpret_cast<void**>(&(qp.qp)));
2837 0 : RPT_INPUT_ERR(
2838 : ret == ROCE_ENOMEM_RET, "EI0011",
2839 : std::vector<std::string>({"memory_size"}), // A3是当ROCE_ENOMEM_RET才上报EI0011,内存大小取决于qp深度配置
2840 : std::vector<std::string>({"262144~3145728"}));
2841 0 : CHK_PRT_RET(
2842 : ret != 0,
2843 : HCCL_ERROR(
2844 : "[Create][NormalQp]errNo[0x%016llx] RaNormalQpCreate fail. return[%d], params: rdmaHandle[%p], context[%p]",
2845 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, rdmaHandle, qp.context),
2846 : HCCL_E_NETWORK);
2847 0 : return HCCL_SUCCESS;
2848 : }
2849 :
2850 0 : HcclResult HrtRaNormalQpDestroy(QpHandle qpHandle)
2851 : {
2852 0 : CHK_PTR_NULL(qpHandle);
2853 0 : HCCL_INFO("[HrtRaNormalQpDestroy] Input params: qpHandle=%p", qpHandle);
2854 0 : s32 ret = RaNormalQpDestroy(qpHandle);
2855 0 : CHK_PRT_RET(
2856 : ret != 0,
2857 : HCCL_ERROR(
2858 : "[Destroy][NormalQp]errNo[0x%016llx] ra destroy normal qp fail. return[%d], params: rdmaHandle[%p]",
2859 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, qpHandle),
2860 : HCCL_E_NETWORK);
2861 0 : return HCCL_SUCCESS;
2862 : }
2863 :
2864 0 : HcclResult HrtRaNdaQpCreate(
2865 : RdmaHandle rdmaHandle, NdaOps* ndaOps, uint32_t dmaMode, NdaCqInfo* cqInfo, NdaQpInfo* qpInfo, QpHandle* qpHandle)
2866 : {
2867 0 : CHK_PTR_NULL(rdmaHandle);
2868 0 : CHK_PTR_NULL(ndaOps);
2869 0 : HCCL_INFO("[HrtRaNdaQpCreate] Input params: rdmaHandle=%p dmaMode=%u", rdmaHandle, dmaMode);
2870 :
2871 : struct ibv_qp_init_attr ibQpAttr;
2872 0 : CHK_SAFETY_FUNC_RET(memset_s(&ibQpAttr, sizeof(ibv_qp_init_attr), 0, sizeof(ibv_qp_init_attr)));
2873 0 : ibQpAttr.qp_context = nullptr;
2874 0 : ibQpAttr.send_cq = cqInfo->cq;
2875 0 : ibQpAttr.recv_cq = cqInfo->cq;
2876 0 : ibQpAttr.srq = nullptr;
2877 0 : ibQpAttr.qp_type = IBV_QPT_RC;
2878 0 : ibQpAttr.cap.max_inline_data = MAX_INLINE_DATA;
2879 0 : ibQpAttr.cap.max_send_wr = MAX_WR_NUM;
2880 0 : ibQpAttr.cap.max_send_sge = MAX_SEND_SGE_NUM;
2881 0 : ibQpAttr.cap.max_recv_wr = MAX_WR_NUM;
2882 0 : ibQpAttr.cap.max_recv_sge = MAX_RECV_SGE_NUM;
2883 :
2884 : struct NdaQpInitAttr qpAttr;
2885 0 : qpAttr.attr = ibQpAttr;
2886 0 : qpAttr.qpCapFlag = 0;
2887 0 : qpAttr.dmaMode = dmaMode;
2888 0 : qpAttr.ops = ndaOps;
2889 :
2890 0 : s32 ret = RaNdaQpCreate(rdmaHandle, &qpAttr, qpInfo, qpHandle);
2891 0 : CHK_PRT_RET(
2892 : ret != 0 || qpInfo == nullptr || qpHandle == nullptr,
2893 : HCCL_ERROR(
2894 : "[Create][NdaQp]errNo[0x%016llx] RaNdaQpCreate fail. return[%d], "
2895 : "params: rdmaHandle[%p] dmaMode[%u]",
2896 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, rdmaHandle, dmaMode),
2897 : HCCL_E_NETWORK);
2898 0 : return HCCL_SUCCESS;
2899 : }
2900 :
2901 2 : HcclResult HrtRaNdaCqCreate(
2902 : RdmaHandle rdmaHandle, NdaOps* ndaOps, uint32_t dmaMode, uint32_t cqAttrFlags, NdaCqInfo* cqInfo,
2903 : CqHandle* cqHandle)
2904 : {
2905 2 : CHK_PTR_NULL(rdmaHandle);
2906 2 : CHK_PTR_NULL(ndaOps);
2907 6 : HCCL_INFO(
2908 : "[HrtRaNdaCqCreate] Input params: rdmaHandle=%p dmaMode=%u cqAttrFlags=%u", rdmaHandle, dmaMode, cqAttrFlags);
2909 :
2910 : struct ibv_cq_init_attr_ex ibCqAttr;
2911 2 : CHK_SAFETY_FUNC_RET(memset_s(&ibCqAttr, sizeof(ibv_cq_init_attr_ex), 0, sizeof(ibv_cq_init_attr_ex)));
2912 2 : if (dmaMode == QBUF_DMA_MODE_INDEP_UB) {
2913 1 : ibCqAttr.cqe = NDA_CQ_DEPTH_FOR_UBNIC;
2914 : } else {
2915 1 : ibCqAttr.cqe = NDA_CQ_DEPTH_FOR_XSCDV;
2916 : }
2917 2 : ibCqAttr.cq_context = nullptr;
2918 2 : ibCqAttr.channel = nullptr;
2919 2 : ibCqAttr.comp_vector = 0;
2920 2 : ibCqAttr.wc_flags = 0;
2921 2 : ibCqAttr.comp_mask = 0;
2922 2 : ibCqAttr.flags = cqAttrFlags;
2923 :
2924 : struct NdaCqInitAttr cqAttr;
2925 2 : cqAttr.attr = ibCqAttr;
2926 2 : cqAttr.cqCapFlag = 0;
2927 2 : cqAttr.dmaMode = dmaMode;
2928 2 : cqAttr.ops = ndaOps;
2929 :
2930 2 : s32 ret = RaNdaCqCreate(rdmaHandle, &cqAttr, cqInfo, cqHandle);
2931 2 : CHK_PRT_RET(
2932 : ret != 0 || cqInfo == nullptr || cqHandle == nullptr,
2933 : HCCL_ERROR(
2934 : "[Create][NdaCq]errNo[0x%016llx] RaNdaCqCreate fail. return[%d], "
2935 : "params: rdmaHandle[%p] dmaMode[%u]",
2936 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, rdmaHandle, dmaMode),
2937 : HCCL_E_NETWORK);
2938 2 : return HCCL_SUCCESS;
2939 : }
2940 :
2941 0 : HcclResult HrtRaNdaCqDestroy(RdmaHandle rdmaHandle, CqHandle cqHandle)
2942 : {
2943 0 : CHK_PTR_NULL(rdmaHandle);
2944 0 : CHK_PTR_NULL(cqHandle);
2945 0 : HCCL_INFO("[HrtRaNdaCqDestroy] Input params: rdmaHandle=%p cqHandle=%p", rdmaHandle, cqHandle);
2946 :
2947 0 : s32 ret = RaNdaCqDestroy(rdmaHandle, cqHandle);
2948 0 : CHK_PRT_RET(
2949 : ret != 0,
2950 : HCCL_ERROR(
2951 : "[RaNdaCqDestroy] errNo[0x%016llx] RaNdaCqDestroy failed, call interface error. "
2952 : "return[%d], params: rdmaHandle[%p] cqHandle[%p]",
2953 : HCCL_ERROR_CODE(HCCL_E_NETWORK), ret, rdmaHandle, cqHandle),
2954 : HCCL_E_NETWORK);
2955 0 : return HCCL_SUCCESS;
2956 : }
2957 :
2958 : HcclResult
2959 0 : RaBatchQueryJettyStatus(const std::vector<JettyHandle>& jettyHandles, std::vector<JettyStatus>& jettyAttrs, u32& num)
2960 : {
2961 0 : if (jettyHandles.size() != num) {
2962 0 : HCCL_ERROR("jettyHandles size[%zu] not equal to num[%u]", jettyHandles.size(), num);
2963 0 : return HCCL_E_PARA;
2964 : }
2965 0 : std::vector<struct JettyAttr> raJettyAttrs(MAX_JETTY_QUERY_NUM);
2966 0 : void* qp_handle[jettyHandles.size()];
2967 0 : for (size_t i = 0; i < jettyHandles.size(); ++i) {
2968 0 : qp_handle[i] = reinterpret_cast<void*>(jettyHandles[i]);
2969 : }
2970 0 : auto ret = RaCtxQpQueryBatch(qp_handle, raJettyAttrs.data(), &num);
2971 0 : if (ret != 0) {
2972 0 : HCCL_ERROR("RaBatchQueryJettyAttr failed.");
2973 0 : return HCCL_E_NETWORK;
2974 : }
2975 0 : if (num != jettyHandles.size()) {
2976 0 : HCCL_ERROR("jettyAttrs num[%zu] not equal to input jettyHandles size[%zu]", num, jettyHandles.size());
2977 0 : return HCCL_E_PARA;
2978 : }
2979 :
2980 0 : for (u32 i = 0; i < num; i++) {
2981 0 : JettyStatus jettyStatus = static_cast<JettyStatus::Value>(static_cast<int>(raJettyAttrs[i].state));
2982 0 : jettyAttrs.push_back(jettyStatus);
2983 : }
2984 0 : return HCCL_SUCCESS;
2985 0 : }
2986 :
2987 0 : HcclResult RaGetAuxInfo(const RdmaHandle rdmaHandle, AuxInfoIn auxInfoIn, AuxInfoOut& auxInfoOut)
2988 : {
2989 : HccpAuxInfoIn in;
2990 0 : in.type = static_cast<HccpAuxInfoInType>(static_cast<int>(auxInfoIn.auxInfoInType));
2991 0 : if (auxInfoIn.auxInfoInType == AuxInfoInType::AUX_INFO_IN_TYPE_CQE) {
2992 0 : in.cqe.status = auxInfoIn.cqe.status;
2993 0 : in.cqe.sR = auxInfoIn.cqe.sR;
2994 0 : } else if (auxInfoIn.auxInfoInType == AuxInfoInType::AUX_INFO_IN_TYPE_AE) {
2995 0 : in.ae.eventType = auxInfoIn.ae.eventType;
2996 : }
2997 :
2998 : HccpAuxInfoOut out;
2999 0 : auto ret = RaCtxGetAuxInfo(rdmaHandle, &in, &out);
3000 0 : if (ret != 0) {
3001 0 : HCCL_ERROR("RaGetAuxInfo failed.");
3002 0 : return HCCL_E_NETWORK;
3003 : }
3004 :
3005 0 : auxInfoOut.auxInfoNum = out.auxInfoNum;
3006 0 : for (uint32_t i = 0; i < out.auxInfoNum; i++) {
3007 0 : auxInfoOut.auxInfoTypes[i] = out.auxInfoType[i];
3008 0 : auxInfoOut.auxInfoValues[i] = out.auxInfoValue[i];
3009 : }
3010 0 : return HCCL_SUCCESS;
3011 : }
3012 :
3013 7 : HcclResult HrtRaCtxQpDestoryBatch(
3014 : const RdmaHandle handle, const std::unordered_set<JettyHandle>& jettyHandles,
3015 : std::vector<JettyHandle>& failJettyHandles)
3016 : {
3017 7 : std::vector<void*> qp_handle;
3018 7 : failJettyHandles.clear();
3019 21 : for (auto jettyHandle : jettyHandles) {
3020 14 : qp_handle.push_back(reinterpret_cast<void*>(jettyHandle));
3021 : }
3022 7 : unsigned int delNum = min(qp_handle.size(), static_cast<size_t>(MAX_DELETE_JETTY_NUMS));
3023 7 : std::vector<void*> del_qp_handle;
3024 : while (true) {
3025 8 : void* raReqHandle = nullptr;
3026 8 : delNum = min(qp_handle.size(), static_cast<size_t>(MAX_DELETE_JETTY_NUMS));
3027 8 : del_qp_handle.assign(qp_handle.begin(), qp_handle.begin() + delNum);
3028 8 : auto ret = RaCtxQpDestroyBatchAsync(handle, del_qp_handle.data(), &delNum, &raReqHandle);
3029 8 : if (ret != 0) {
3030 3 : HCCL_ERROR("[%s] failed, ret is [%d].", __func__, ret);
3031 3 : return HCCL_E_INTERNAL;
3032 : }
3033 :
3034 7 : RequestHandle reqHandle = reinterpret_cast<RequestHandle>(raReqHandle);
3035 7 : auto startTime = std::chrono::steady_clock::now();
3036 7 : constexpr uint32_t pollTimeoutMs = 10000; // 轮询超时时间10s
3037 7 : auto waitPollTimeOutMs = std::chrono::milliseconds(pollTimeoutMs);
3038 : while (true) {
3039 5473528 : if ((std::chrono::steady_clock::now() - startTime) >= waitPollTimeOutMs) {
3040 3 : HCCL_ERROR(
3041 : "[%s]poll timeout, originalJettyCount[%zu], undeleteJettyCount[%zu].", __func__,
3042 : jettyHandles.size(), failJettyHandles.size());
3043 1 : return HCCL_E_TIMEOUT;
3044 : }
3045 5473527 : ReqHandleResult result = ReqHandleResult::INVALID_PARA;
3046 5473527 : TRY_CATCH_RETURN(result = HrtRaGetAsyncReqResult(reqHandle));
3047 5473527 : if (result == ReqHandleResult::NOT_COMPLETED) {
3048 5473521 : continue;
3049 6 : } else if (result == ReqHandleResult::COMPLETED) {
3050 6 : break;
3051 : } else {
3052 0 : HCCL_ERROR("[%s] failed, result[%s] is unexpected.", __func__, result.Describe().c_str());
3053 0 : return HCCL_E_INTERNAL;
3054 : }
3055 5473521 : }
3056 :
3057 : // 检查是否删除完成
3058 6 : if (delNum > del_qp_handle.size()) {
3059 3 : HCCL_ERROR(
3060 : "[%s] run RaCtxQpDestroyBatchAsync error, del jetty num[%u] greater than all jetty num[%zu].", __func__,
3061 : delNum, del_qp_handle.size());
3062 1 : return HCCL_E_INTERNAL;
3063 5 : } else if (del_qp_handle.size() == delNum) {
3064 3 : qp_handle.erase(qp_handle.begin(), qp_handle.begin() + delNum);
3065 : } else {
3066 2 : failJettyHandles.push_back(reinterpret_cast<JettyHandle>(del_qp_handle[delNum]));
3067 2 : qp_handle.erase(qp_handle.begin(), qp_handle.begin() + delNum + 1);
3068 : }
3069 5 : if (qp_handle.size() == 0) {
3070 4 : break;
3071 : }
3072 1 : }
3073 12 : HCCL_INFO(
3074 : "[%s] run success, originalJettyCount[%zu], undeleteJettyCount[%zu].", __func__, jettyHandles.size(),
3075 : failJettyHandles.size());
3076 4 : return HCCL_SUCCESS;
3077 7 : }
3078 :
3079 : struct ccu_mem_info {
3080 : unsigned int long long mem_va;
3081 : unsigned int mem_size;
3082 : unsigned int resv[1];
3083 : };
3084 :
3085 : struct ccu_mem_rsp {
3086 : unsigned int die_id;
3087 : unsigned int num;
3088 : struct ccu_mem_info list[64U];
3089 : };
3090 :
3091 6 : void HrtSetMemInfoList(struct CcuMemInfo* memInfoList, uint32_t count, struct ccu_mem_info* recvMemList)
3092 : {
3093 114 : for (size_t i = 0; i < count; ++i) {
3094 108 : memInfoList[i].memVa = recvMemList[i].mem_va;
3095 108 : memInfoList[i].memSize = recvMemList[i].mem_size;
3096 : }
3097 6 : }
3098 :
3099 6 : HcclResult HrtGetCcuMemInfo(
3100 : void* tlv_handle, uint32_t udieIdx, uint64_t memTypeBitmap, struct CcuMemInfo* memInfoList, uint32_t count)
3101 : {
3102 6 : s32 ret = 0;
3103 6 : u32 tlv_module_type = TLV_MODULE_TYPE_CCU;
3104 :
3105 6 : struct TlvMsg send_msg = {};
3106 6 : struct TlvMsg recv_msg = {};
3107 : // 使用unique_ptr管理动态分配的内存，实现RAII
3108 6 : auto send_data = std::make_unique<char[]>(sizeof(CcuMemReq));
3109 6 : auto recv_data = std::make_unique<char[]>(sizeof(ccu_mem_rsp));
3110 :
3111 : // 初始化请求消息
3112 6 : send_msg.type = MSG_TYPE_CCU_GET_MEM_INFO;
3113 6 : send_msg.length = sizeof(CcuMemReq);
3114 6 : send_msg.data = send_data.get();
3115 :
3116 6 : auto req = reinterpret_cast<CcuMemReq*>(send_msg.data);
3117 6 : req->udieIdx = udieIdx;
3118 6 : req->memTypeBitmap = memTypeBitmap;
3119 :
3120 : // 初始化响应消息
3121 6 : recv_msg.type = 0;
3122 6 : recv_msg.length = sizeof(ccu_mem_rsp);
3123 6 : recv_msg.data = recv_data.get();
3124 :
3125 6 : auto rsp = reinterpret_cast<ccu_mem_rsp*>(recv_msg.data);
3126 6 : rsp->die_id = 0;
3127 6 : rsp->num = 0;
3128 18 : std::fill(std::begin(rsp->list), std::end(rsp->list), ccu_mem_info{});
3129 :
3130 6 : ret = RaTlvRequest(tlv_handle, tlv_module_type, &send_msg, &recv_msg);
3131 6 : if (ret != 0) {
3132 0 : if (ret == RA_TLV_REQUEST_UNAVAIL) {
3133 0 : HCCL_WARNING("[HrtGetCcuMemInfo]ra tlv request UNAVAIL. return: ret[%d]", ret);
3134 0 : return HCCL_E_UNAVAIL;
3135 : }
3136 0 : HCCL_ERROR(
3137 : "[Request][RaTlv]errNo[0x%016llx] ra tlv request fail. return: ret[%d], module type[%u], message type[%u]",
3138 : HCCL_ERROR_CODE(HcclResult::HCCL_E_NETWORK), ret, tlv_module_type, send_msg.type);
3139 0 : throw NetworkApiException(StringFormat("call ra_tlv_request failed"));
3140 : }
3141 6 : HrtSetMemInfoList(memInfoList, count, rsp->list);
3142 18 : HCCL_INFO("tlv request success, tlv module type[%u], message type[%u]", tlv_module_type, send_msg.type);
3143 6 : return HCCL_SUCCESS;
3144 6 : }
3145 :
3146 4 : HcclResult HrtRaGetEidByIp(RdmaHandle handle, const vector<IpAddress>& ipV4AddrList, vector<IpAddress>& eidAddrList)
3147 : {
3148 12 : HCCL_INFO("[HrtRaGetEidByIp] begain, ipV4AddrList size=%zu", ipV4AddrList.size());
3149 4 : size_t ipV4AddrListSize = ipV4AddrList.size();
3150 4 : unsigned int num = ipV4AddrListSize;
3151 7 : IpInfo ipInfoList[num] = {};
3152 7 : for (size_t i = 0; i < num; i++) {
3153 3 : auto ipAddress = ipV4AddrList.at(i);
3154 9 : HCCL_INFO("[HrtRaGetEidByIp] ipV4AddrList[%d][%s]", i, ipAddress.Describe().c_str());
3155 3 : ipInfoList[i].family = ipAddress.GetFamily();
3156 3 : ipInfoList[i].ip = IpAddressToHccpIpAddr(ipAddress);
3157 : }
3158 :
3159 7 : union HccpEid eidList[num] = {};
3160 4 : s32 ret = RaGetEidByIp(handle, ipInfoList, eidList, &num);
3161 4 : if (ret != 0) {
3162 3 : HCCL_WARNING("call RaGetEidByIp failed, error code =%d.", ret);
3163 1 : return HCCL_E_INTERNAL;
3164 : }
3165 :
3166 3 : if (num != ipV4AddrList.size()) {
3167 3 : HCCL_ERROR(
3168 : "call RaGetEidByIp failed, The number of ipInfoList and eidList is inconsistent, "
3169 : "ipV4AddrList size =%zu, eidList size =%u",
3170 : ipV4AddrList.size(), num);
3171 1 : return HCCL_E_INTERNAL;
3172 : }
3173 :
3174 3 : for (unsigned int i = 0; i < num; i++) {
3175 1 : IpAddress eidAddr = HccpEidToIpAddress(eidList[i]);
3176 1 : eidAddrList.push_back(eidAddr);
3177 : }
3178 6 : HCCL_INFO("[HrtRaGetEidByIp] success, eidAddrList size=%zu", eidAddrList.size());
3179 2 : return HCCL_SUCCESS;
3180 4 : }
3181 :
3182 3 : HcclResult WaitRequestResult(void* raReqHandle, RequestHandle& reqHandle)
3183 : {
3184 3 : reqHandle = reinterpret_cast<RequestHandle>(raReqHandle);
3185 3 : auto startTime = std::chrono::steady_clock::now();
3186 3 : constexpr uint32_t pollTimeoutMs = 10000; // 轮询超时时间
3187 3 : auto waitPollTimeOutMs = std::chrono::milliseconds(pollTimeoutMs);
3188 : while (true) {
3189 3 : if ((std::chrono::steady_clock::now() - startTime) >= waitPollTimeOutMs) {
3190 0 : HCCL_ERROR("[WaitRequestResult] poll timeout.");
3191 1 : return HCCL_E_TIMEOUT; // 超时报错
3192 : }
3193 :
3194 3 : ReqHandleResult result = ReqHandleResult::INVALID_PARA;
3195 3 : TRY_CATCH_RETURN(result = HrtRaGetAsyncReqResult(reqHandle));
3196 :
3197 : // 结果判断
3198 3 : if (result == ReqHandleResult::NOT_COMPLETED) {
3199 0 : continue;
3200 3 : } else if (result == ReqHandleResult::COMPLETED) {
3201 2 : break;
3202 : } else {
3203 3 : HCCL_ERROR("[WaitRequestResult] failed, result[%s] is unexpected.", result.Describe().c_str());
3204 1 : return HCCL_E_INTERNAL;
3205 : }
3206 0 : }
3207 :
3208 2 : return HCCL_SUCCESS;
3209 : }
3210 :
3211 0 : HcclResult HrtRaGetTpAttr(RdmaHandle handle, uint64_t tpHandle, uint32_t& attrBitmap, TpAttr& attr)
3212 : {
3213 0 : const s32 ret = RaCtxGetTpAttr(handle, tpHandle, &attrBitmap, &attr);
3214 0 : if (ret != 0) {
3215 0 : HCCL_ERROR(
3216 : "[HrtRaGetTpAttr] RaCtxGetTpAttr failed ret[%d] tpHandle[%llu] attrBitmap[0x%x].", ret, tpHandle,
3217 : attrBitmap);
3218 0 : return HCCL_E_NETWORK;
3219 : }
3220 0 : return HCCL_SUCCESS;
3221 : }
3222 :
3223 0 : HcclResult HrtRaSetTpAttr(RdmaHandle handle, uint64_t tpHandle, uint32_t attrBitmap, TpAttr& attr)
3224 : {
3225 0 : const s32 ret = RaCtxSetTpAttr(handle, tpHandle, attrBitmap, &attr);
3226 0 : if (ret != 0) {
3227 0 : HCCL_ERROR(
3228 : "[HrtRaSetTpAttr] RaCtxSetTpAttr failed ret[%d] tpHandle[%llu] attrBitmap[0x%x].", ret, tpHandle,
3229 : attrBitmap);
3230 0 : return HCCL_E_NETWORK;
3231 : }
3232 0 : return HCCL_SUCCESS;
3233 : }
3234 :
3235 22 : bool HrtRaSupportsGetTpAttr(u32 phyId)
3236 : {
3237 22 : u32 tpAttrVersion = 0;
3238 22 : const s32 ret = RaGetInterfaceVersion(phyId, GET_TP_ATTR_OPCODE, &tpAttrVersion);
3239 44 : return (ret == 0 && tpAttrVersion >= GET_TP_ATTR_VERSION);
3240 : }
3241 :
3242 : HcclResult
3243 3 : HrtRaSetTpAttrAsync(RdmaHandle handle, uint64_t tpHandle, uint32_t attrBitmap, TpAttr& attr, RequestHandle& reqHandle)
3244 : {
3245 9 : HCCL_INFO("[HrtRaSetTpAttrAsync] begain, reqHandle[%llu]", reqHandle);
3246 3 : void* raReqHandle = nullptr;
3247 3 : s32 ret = RaSetTpAttrAsync(handle, tpHandle, attrBitmap, &attr, &raReqHandle);
3248 3 : if (ret != 0) {
3249 1 : string msg = StringFormat("call RaSetTpAttrAsync failed, error code =%d.", ret);
3250 1 : THROW<NetworkApiException>(msg);
3251 1 : }
3252 :
3253 5 : CHK_RET(WaitRequestResult(raReqHandle, reqHandle));
3254 3 : HCCL_INFO("[HrtRaSetTpAttrAsync] success, reqHandle[%llu]", reqHandle);
3255 1 : return HCCL_SUCCESS;
3256 : }
3257 :
3258 5 : HcclResult HrtRaGetTpAttrAsync(
3259 : u32 phyId, RdmaHandle handle, uint64_t tpHandle, uint32_t& attrBitmap, TpAttr& attr, RequestHandle& reqHandle)
3260 : {
3261 15 : HCCL_INFO("[HrtRaGetTpAttrAsync] begain, reqHandle[%llu]", reqHandle);
3262 5 : if (!HrtRaSupportsGetTpAttr(phyId)) {
3263 9 : HCCL_ERROR(
3264 : "this package does not support RaGetTpAttrAsync for device, please change new package, phyId=%u.", phyId);
3265 3 : return HCCL_E_NOT_SUPPORT;
3266 : }
3267 2 : void* raReqHandle = nullptr;
3268 2 : s32 ret = RaGetTpAttrAsync(handle, tpHandle, &attrBitmap, &attr, &raReqHandle);
3269 2 : if (ret != 0) {
3270 1 : string msg = StringFormat("call RaGetTpAttrAsync failed, error code =%d.", ret);
3271 1 : THROW<NetworkApiException>(msg);
3272 1 : }
3273 :
3274 1 : CHK_RET(WaitRequestResult(raReqHandle, reqHandle));
3275 3 : HCCL_INFO("[HrtRaGetTpAttrAsync] success, reqHandle[%llu]", reqHandle);
3276 1 : return HCCL_SUCCESS;
3277 : }
3278 :
3279 10 : HcclResult HrtRaStartGetTpAttrAsync(
3280 : RdmaHandle handle, uint64_t tpHandle, uint32_t& attrBitmap, TpAttr& attr, RequestHandle& reqHandle)
3281 : {
3282 10 : void* raReqHandle = nullptr;
3283 10 : const s32 ret = RaGetTpAttrAsync(handle, tpHandle, &attrBitmap, &attr, &raReqHandle);
3284 10 : if (ret != 0 || raReqHandle == nullptr) {
3285 0 : HCCL_ERROR(
3286 : "[HrtRaStartGetTpAttrAsync] RaGetTpAttrAsync failed ret[%d] raReqHandle[%p] tpHandle[%llu].", ret,
3287 : raReqHandle, tpHandle);
3288 0 : return HCCL_E_NETWORK;
3289 : }
3290 10 : reqHandle = reinterpret_cast<RequestHandle>(raReqHandle);
3291 10 : return HCCL_SUCCESS;
3292 : }
3293 :
3294 3 : HcclResult HrtGetUboeFlagEnable(const u32 devPhyId)
3295 : {
3296 3 : u32 uboeVersion = 0;
3297 3 : s32 versionRet = RaGetInterfaceVersion(devPhyId, GET_UBOE_FLAG_ENABLE_OPCODE, &uboeVersion);
3298 6 : CHK_PRT_RET(
3299 : versionRet != 0,
3300 : HCCL_ERROR("[%s] RaGetInterfaceVersion failed, devPhyId=%u, versionRet=%d", __func__, devPhyId, versionRet),
3301 : HCCL_E_INTERNAL);
3302 5 : CHK_PRT_RET(
3303 : uboeVersion < GET_UBOE_FLAG_ENABLE_VERSION,
3304 : HCCL_ERROR(
3305 : "[%s] this package does not support to get uboe flag, "
3306 : "please change new package. uboeVersion[%u].",
3307 : __func__, uboeVersion),
3308 : HCCL_E_NOT_SUPPORT);
3309 1 : return HCCL_SUCCESS;
3310 : }
3311 :
3312 : } // namespace Hccl
|
