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