Line data Source code
1 : /**
2 : * Copyright (c) 2026 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 : #include <cstring>
11 : #include <memory>
12 : #include <vector>
13 :
14 : #include "hcomm_c_adpt.h"
15 : #include "hcomm_c_adpt_common.h"
16 : #include "log.h"
17 : #include "endpoint.h"
18 : #include "endpoint_monitor.h"
19 : #include "../hcomm_res_mgr.h"
20 : #include "hcomm_result_defs.h"
21 : #include "param_check_pub.h"
22 : #include "exception_handler.h"
23 : #include "hcom_common.h"
24 : #include "hcomm_res_defs.h"
25 : #include "channel_config.h"
26 : #include "shared_jetty_mgr.h"
27 : #include "endpoint.h"
28 : #include "builtin_endpoint_ops.h"
29 : #include "nic_plugin_holder.h"
30 : #include "nic_plugin_manager.h"
31 : #include "hcomm_adapter_hccp.h"
32 : #include "eid_info_mgr.h"
33 : #include "hccp_hdc_manager.h"
34 : #include "orion_adpt_utils.h"
35 : #include "rdma_handle_manager.h"
36 : #include "adapter_rts_common.h"
37 :
38 : using namespace hcomm;
39 :
40 : namespace {
41 5 : HcclResult GetDeviceEidInfos(int32_t deviceLogicId, uint32_t devicePhyId, std::vector<DevEidInfo>& eidInfos)
42 : {
43 5 : Hccl::HccpHdcManager::GetInstance().Init(static_cast<uint32_t>(deviceLogicId));
44 5 : CHK_RET(EidInfoMgr::GetInstance(devicePhyId).GetEidInfos(eidInfos));
45 4 : CHK_PRT_RET(
46 : eidInfos.empty(),
47 : HCCL_ERROR(
48 : "[%s] no endpoint EID found, deviceLogicId[%d], devicePhyId[%u].", __func__, deviceLogicId, devicePhyId),
49 : HCCL_E_NOT_FOUND);
50 3 : return HCCL_SUCCESS;
51 : }
52 :
53 3 : HcclResult GetEidAddress(const DevEidInfo& eidInfo, Hccl::IpAddress& eidAddress)
54 : {
55 3 : CHK_RET(CommAddrToIpAddress(eidInfo.commAddr, eidAddress));
56 3 : CHK_PRT_RET(
57 : eidAddress.IsInvalid(), HCCL_ERROR("[%s] EID address is invalid, eidIndex[%u].", __func__, eidInfo.eidIndex),
58 : HCCL_E_PARA);
59 3 : return HCCL_SUCCESS;
60 : }
61 :
62 3 : HcclResult FillEidCommAddr(const Hccl::IpAddress& eidAddress, CommAddr& commAddr)
63 : {
64 3 : commAddr.type = COMM_ADDR_TYPE_EID;
65 3 : const Hccl::Eid& eid = eidAddress.GetEid();
66 3 : CHK_SAFETY_FUNC_RET(memcpy_s(commAddr.eid, sizeof(commAddr.eid), eid.raw, sizeof(eid.raw)));
67 3 : return HCCL_SUCCESS;
68 : }
69 :
70 3 : HcclResult FillEndpointDesc(uint32_t devicePhyId, const DevEidInfo& eidInfo, EndpointDesc& endpointDesc)
71 : {
72 3 : Hccl::IpAddress eidAddress{};
73 3 : CHK_RET(GetEidAddress(eidInfo, eidAddress));
74 :
75 3 : endpointDesc.loc.locType = ENDPOINT_LOC_TYPE_DEVICE;
76 3 : endpointDesc.loc.device.devPhyId = devicePhyId;
77 :
78 3 : CommAddr eidCommAddr{};
79 3 : CHK_RET(FillEidCommAddr(eidAddress, eidCommAddr));
80 :
81 3 : auto& rdmaHandleMgr = Hccl::RdmaHandleManager::GetInstance();
82 : Hccl::RdmaHandle rdmaHandle
83 3 : = rdmaHandleMgr.GetByAddr(devicePhyId, Hccl::LinkProtoType::UB, eidAddress, Hccl::PortDeploymentType::DEV_NET);
84 3 : CHK_PTR_NULL(rdmaHandle);
85 :
86 3 : if (HccpCheckUboeSupported(eidInfo.devFeature)) {
87 1 : endpointDesc.protocol = COMM_PROTOCOL_UBOE;
88 1 : return HccpGetIpByEid(rdmaHandle, eidCommAddr, endpointDesc.commAddr);
89 : }
90 :
91 2 : endpointDesc.commAddr = eidCommAddr;
92 2 : bool ctpEnable = false;
93 2 : CHK_RET(HccpGetCtpEnable(rdmaHandle, ctpEnable));
94 2 : endpointDesc.protocol = ctpEnable ? COMM_PROTOCOL_UBC_CTP : COMM_PROTOCOL_UBG;
95 2 : return HCCL_SUCCESS;
96 : }
97 :
98 91 : HcclResult ValidateEndpointDesc(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
99 : {
100 91 : CHK_PTR_NULL(endpoint);
101 90 : CHK_PTR_NULL(endpointHandle);
102 89 : if (endpoint->loc.locType != ENDPOINT_LOC_TYPE_DEVICE && endpoint->loc.locType != ENDPOINT_LOC_TYPE_HOST) {
103 0 : HCCL_ERROR(
104 : "[%s] Only support END_POINT_LOCATION_DEVICE AND END_POINT_LOCATION_HOST, but "
105 : "endpoint->loc.locType is %d",
106 : __func__, endpoint->loc.locType);
107 0 : return HCCL_E_PARA;
108 : }
109 89 : return HCCL_SUCCESS;
110 : }
111 :
112 78 : HcclResult RegisterDeviceEndpointMonitorIfNeeded(const EndpointDesc* endpoint, EndpointHandle handle)
113 : {
114 78 : if ((endpoint->loc.locType != ENDPOINT_LOC_TYPE_DEVICE)
115 61 : || ((endpoint->protocol != COMM_PROTOCOL_UBC_CTP) && (endpoint->protocol != COMM_PROTOCOL_UBC_TP))) {
116 61 : return HCCL_SUCCESS;
117 : }
118 :
119 17 : s32 devLogicIdSigned = HcclGetThreadDeviceId();
120 17 : CHK_PRT_RET(
121 : devLogicIdSigned < 0, HCCL_ERROR("[%s] HcclGetThreadDeviceId failed, ret[%d]", __func__, devLogicIdSigned),
122 : HCCL_E_INTERNAL);
123 17 : EndpointMonitor::GetInstance(devLogicIdSigned).RegisterToEndpointMonitor(devLogicIdSigned, handle);
124 17 : return HCCL_SUCCESS;
125 : }
126 :
127 4 : HcommResult CreatePluginEndpointHolder(
128 : const EndpointDesc* endpoint, const NicPluginEntry* pluginEntry, EndpointHandle* endpointHandle)
129 : {
130 4 : CHK_PTR_NULL(endpoint);
131 4 : CHK_PTR_NULL(pluginEntry);
132 4 : CHK_PTR_NULL(endpointHandle);
133 4 : void* pluginCtx = nullptr;
134 4 : HcommNicEndpointOps* pluginOps = nullptr;
135 4 : HcommResult ret = HCCL_SUCCESS;
136 4 : ret = static_cast<HcommResult>(pluginEntry->createEndpoint(endpoint, &pluginCtx, &pluginOps));
137 4 : CHK_PRT_RET(
138 : ret != HCCL_SUCCESS,
139 : HCCL_ERROR(
140 : "[NicPlugin][%s] plugin createEndpoint failed, ret[%d], protocol[%d].", __func__, ret, endpoint->protocol),
141 : ret);
142 :
143 4 : CHK_PRT_RET(
144 : !ValidateEndpointOps(pluginOps),
145 : HCCL_ERROR(
146 : "[NicPlugin][%s] plugin endpoint ops validation failed, protocol[%d].", __func__, endpoint->protocol),
147 : HCCL_E_PARA);
148 :
149 : // 确保HcommNicEndpointOps各接口非空实现，后续调用处无需校验
150 4 : HcommNicEndpointOps* pluginHolderOps = nullptr;
151 4 : ret = FillDefaultEndpointOps(pluginOps, &pluginHolderOps);
152 4 : CHK_PRT_RET(
153 : ret != HCCL_SUCCESS,
154 : HCCL_ERROR(
155 : "[NicPlugin][%s] FillDefaultEndpointOps failed, ret[%d], protocol[%d].", __func__, ret, endpoint->protocol),
156 : ret);
157 :
158 4 : ret = static_cast<HcommResult>(pluginHolderOps->init(pluginCtx));
159 4 : if (ret != HCCL_SUCCESS) {
160 0 : int32_t destroyRet = pluginHolderOps->destroy(pluginCtx);
161 0 : if (destroyRet != HCCL_SUCCESS) {
162 0 : HCCL_WARNING("[%s] plugin endpoint destroy failed after init failure, ret[%d].", __func__, destroyRet);
163 : }
164 0 : delete pluginHolderOps;
165 0 : HCCL_ERROR("[NicPlugin][%s] plugin endpoint init failed, ret[%d].", __func__, ret);
166 0 : return ret;
167 : }
168 :
169 4 : auto holder = std::make_unique<PluginEndpointHolder>(*endpoint, pluginEntry);
170 4 : holder->SetNicEndpointCtx(pluginHolderOps, pluginCtx);
171 4 : const EndpointHandle handle = reinterpret_cast<EndpointHandle>(holder.get());
172 4 : EXCEPTION_CATCH(GetEndpointMap().AddEndpoint(handle, std::move(holder)), return HCCL_E_INTERNAL);
173 4 : *endpointHandle = handle;
174 4 : HCCL_INFO(
175 : "[NicPlugin][%s] plugin endpoint created, protocol[%d], handle[%p].", __func__, endpoint->protocol, handle);
176 4 : return HCCL_SUCCESS;
177 4 : }
178 :
179 85 : HcclResult CreateBuiltinEndpoint(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
180 : {
181 85 : CHK_RET(RefreshEndpointContext(*endpoint));
182 85 : std::unique_ptr<Endpoint> endpointPtr = nullptr;
183 85 : HcclResult ret = Endpoint::CreateEndpoint(*endpoint, endpointPtr);
184 85 : if (ret != HCCL_SUCCESS) {
185 2 : HCCL_ERROR("call Endpoint::CreateEndpoint failed");
186 2 : return ret;
187 : }
188 83 : CHK_PTR_NULL(endpointPtr);
189 83 : ret = endpointPtr->Init();
190 83 : if (ret != HCCL_SUCCESS) {
191 5 : HCCL_ERROR("call endpointPtr->Init failed");
192 5 : return ret;
193 : }
194 :
195 78 : endpointPtr->SetNicEndpointCtx(&g_BuiltinEndpointOps, endpointPtr.get());
196 :
197 78 : const EndpointHandle handle = reinterpret_cast<EndpointHandle>(endpointPtr.get());
198 78 : CHK_PTR_NULL(handle);
199 78 : EXCEPTION_CATCH(GetEndpointMap().AddEndpoint(handle, std::move(endpointPtr)), return HCCL_E_INTERNAL);
200 78 : *endpointHandle = handle;
201 78 : CHK_RET(RegisterDeviceEndpointMonitorIfNeeded(endpoint, handle));
202 78 : HCCL_INFO(
203 : "[%s] endpointDesc.protocol [%d] and endpointDesc.loc.locType [%d] create endpointHandle [%p] done.", __func__,
204 : endpoint->protocol, endpoint->loc.locType, handle);
205 78 : return HCCL_SUCCESS;
206 85 : }
207 : } // namespace
208 :
209 5 : HcommResult HcommEndpointGetDescNum(int32_t deviceLogicId, uint32_t* descNum)
210 : {
211 : EXCEPTION_HANDLE_BEGIN
212 9 : CHK_PTR_NULL(descNum);
213 5 : *descNum = 0;
214 5 : HcommResMgr::RegisterDeviceResetCallback();
215 :
216 5 : CHK_PRT_RET(
217 : deviceLogicId < 0, HCCL_ERROR("[%s] deviceLogicId[%d] is invalid.", __func__, deviceLogicId), HCCL_E_PARA);
218 5 : DevType deviceType = DevType::DEV_TYPE_COUNT;
219 5 : CHK_RET(hrtGetDeviceType(deviceType));
220 5 : CHK_PRT_RET(
221 : deviceType != DevType::DEV_TYPE_950,
222 : HCCL_ERROR(
223 : "[%s] endpoint query only supports DEV_TYPE_950, current deviceType[%d].", __func__,
224 : static_cast<int32_t>(deviceType)),
225 : HCCL_E_NOT_SUPPORT);
226 3 : uint32_t devicePhyId = 0;
227 3 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId));
228 3 : std::vector<DevEidInfo> eidInfos;
229 3 : CHK_RET(GetDeviceEidInfos(deviceLogicId, devicePhyId, eidInfos));
230 :
231 1 : *descNum = static_cast<uint32_t>(eidInfos.size());
232 1 : HCCL_INFO(
233 : "[%s] success, deviceLogicId[%d], devicePhyId[%u], descNum[%u].", __func__, deviceLogicId, devicePhyId,
234 : *descNum);
235 3 : EXCEPTION_HANDLE_END
236 1 : return HCCL_SUCCESS;
237 : }
238 :
239 2 : HcommResult HcommEndpointGetDescs(int32_t deviceLogicId, uint32_t* descNum, EndpointDesc* endpointDescs)
240 : {
241 : EXCEPTION_HANDLE_BEGIN
242 2 : CHK_PTR_NULL(descNum);
243 2 : CHK_PTR_NULL(endpointDescs);
244 2 : HcommResMgr::RegisterDeviceResetCallback();
245 :
246 2 : uint32_t devicePhyId = 0;
247 2 : std::vector<DevEidInfo> eidInfos;
248 2 : CHK_PRT_RET(
249 : deviceLogicId < 0, HCCL_ERROR("[%s] deviceLogicId[%d] is invalid.", __func__, deviceLogicId), HCCL_E_PARA);
250 2 : const uint32_t capacity = *descNum;
251 2 : DevType deviceType = DevType::DEV_TYPE_COUNT;
252 2 : CHK_RET(hrtGetDeviceType(deviceType));
253 2 : CHK_PRT_RET(
254 : deviceType != DevType::DEV_TYPE_950,
255 : HCCL_ERROR(
256 : "[%s] endpoint query only supports DEV_TYPE_950, current deviceType[%d].", __func__,
257 : static_cast<int32_t>(deviceType)),
258 : HCCL_E_NOT_SUPPORT);
259 2 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId));
260 2 : CHK_RET(GetDeviceEidInfos(deviceLogicId, devicePhyId, eidInfos));
261 2 : CHK_PRT_RET(
262 : eidInfos.size() > capacity,
263 : HCCL_ERROR(
264 : "[%s] endpointDescs capacity[%u] is smaller than required count[%zu].", __func__, capacity,
265 : eidInfos.size()),
266 : HCCL_E_PARA);
267 :
268 2 : const uint32_t actualNum = static_cast<uint32_t>(eidInfos.size());
269 2 : const HcommResult initRet = EndpointDescInit(endpointDescs, actualNum);
270 2 : CHK_PRT_RET(
271 : initRet != HCOMM_SUCCESS, HCCL_ERROR("[%s] EndpointDescInit failed, ret[%d].", __func__, initRet), initRet);
272 5 : for (uint32_t i = 0; i < actualNum; ++i) {
273 3 : CHK_RET(FillEndpointDesc(devicePhyId, eidInfos[i], endpointDescs[i]));
274 : }
275 :
276 2 : *descNum = actualNum;
277 2 : HCCL_INFO(
278 : "[%s] success, deviceLogicId[%d], devicePhyId[%u], descNum[%u].", __func__, deviceLogicId, devicePhyId,
279 : *descNum);
280 2 : EXCEPTION_HANDLE_END
281 2 : return HCCL_SUCCESS;
282 : }
283 :
284 23 : HcommResult HcommEndpointGet(EndpointHandle endpointHandle, void** endpoint) // 根据endpointHandle返回Endpoint对象指针
285 : {
286 23 : CHK_PTR_NULL(endpoint);
287 :
288 20 : auto it = GetEndpointMap().GetEndpoint(endpointHandle);
289 20 : CHK_PRT_RET(
290 : it == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
291 : HCCL_E_NOT_FOUND);
292 :
293 18 : *endpoint = static_cast<void*>(it);
294 18 : HCCL_INFO(
295 : "[%s] START. endpointHandle[%p] endpoint[%p].", __func__, static_cast<void*>(endpointHandle),
296 : static_cast<void*>(endpoint));
297 18 : return HCCL_SUCCESS;
298 : }
299 :
300 91 : HcommResult HcommEndpointCreate(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
301 : {
302 : EXCEPTION_HANDLE_BEGIN
303 91 : CHK_RET(ValidateEndpointDesc(endpoint, endpointHandle));
304 89 : if (endpoint->loc.locType == ENDPOINT_LOC_TYPE_HOST) {
305 24 : const NicPluginEntry* pluginEntry = FindHostNicPlugin(endpoint->protocol);
306 24 : if (pluginEntry != nullptr) {
307 4 : return CreatePluginEndpointHolder(endpoint, pluginEntry, endpointHandle);
308 : }
309 : }
310 85 : (void)HcommResMgrInit();
311 85 : CHK_RET(CreateBuiltinEndpoint(endpoint, endpointHandle));
312 78 : HcommResMgr::RegisterDeviceResetCallback();
313 0 : EXCEPTION_HANDLE_END
314 78 : return HCCL_SUCCESS;
315 : }
316 :
317 53 : HcommResult HcommEndpointDestroy(EndpointHandle endpointHandle)
318 : {
319 53 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].", __func__, endpointHandle);
320 53 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
321 53 : if (endpoint != nullptr && endpoint->GetNicOps() != nullptr && endpoint->GetNicOps() != &g_BuiltinEndpointOps) {
322 4 : HCCL_INFO("[NicPlugin][%s] destroy plugin endpoint.", __func__);
323 4 : auto ret = GetEndpointMap().RemoveEndpoint(endpointHandle);
324 4 : CHK_PRT_RET(
325 : ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]", __func__, endpointHandle),
326 : HCCL_E_NOT_FOUND);
327 4 : return HCCL_SUCCESS;
328 : }
329 49 : (void)HcommResMgrInit();
330 : // 需校验共享 jetty channel 是否已全部销毁，否则残留 channel 持有的 jetty 引用会在 endpoint 销毁后成为悬空引用。
331 49 : HcclResult jettyRet = hcomm::SharedJettyMgr::GetInstance().CheckEndpointDestroy(endpointHandle);
332 49 : CHK_PRT_RET(
333 : jettyRet != HCCL_SUCCESS,
334 : HCCL_ERROR(
335 : "[%s] cannot destroy endpointHandle[0x%llx], shared jetty channels still exist.", __func__, endpointHandle),
336 : jettyRet);
337 49 : if (endpoint != nullptr) {
338 47 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
339 : }
340 49 : s32 devLogicIdSigned = HcclGetThreadDeviceId();
341 49 : CHK_PRT_RET(
342 : devLogicIdSigned < 0, HCCL_ERROR("[%s] HcclGetThreadDeviceId failed, ret[%d]", __func__, devLogicIdSigned),
343 : HCCL_E_INTERNAL);
344 49 : EndpointMonitor::GetInstance(devLogicIdSigned).RemoveEpHandleFromEndpointMonitor(endpointHandle);
345 49 : auto ret = GetEndpointMap().RemoveEndpoint(endpointHandle);
346 49 : CHK_PRT_RET(
347 : ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]", __func__, endpointHandle),
348 : HCCL_E_NOT_FOUND);
349 47 : return HCCL_SUCCESS;
350 : }
351 :
352 25 : HcommResult HcommEndpointStartListen(EndpointHandle endpointHandle, uint32_t port, HcommEndpointListenConfig* config)
353 : {
354 : (void)config;
355 25 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
356 25 : CHK_PRT_RET(
357 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
358 : HCCL_E_NOT_FOUND);
359 24 : CHK_RET(endpoint->ServerSocketListen(port));
360 24 : return HCCL_SUCCESS;
361 : }
362 :
363 3 : HcommResult HcommEndpointStopListen(EndpointHandle endpointHandle, uint32_t port)
364 : {
365 3 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
366 3 : CHK_PRT_RET(
367 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
368 : HCCL_E_NOT_FOUND);
369 2 : CHK_RET(endpoint->ServerSocketStopListen(port));
370 2 : return HCCL_SUCCESS;
371 : }
372 :
373 8 : HcommResult HcommEndpointGetListenPort(EndpointHandle endpointHandle, uint32_t* port)
374 : {
375 8 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
376 8 : CHK_PRT_RET(
377 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
378 : HCCL_E_NOT_FOUND);
379 5 : return static_cast<HcclResult>(endpoint->GetNicOps()->getListenPort(endpoint->GetNicCtx(), port));
380 : }
381 :
382 : HcommResult
383 1 : HcommEndpointCheckFeature(HcommEndpointFeatureType featureType, const EndpointDesc* endpointDesc, bool* value)
384 : {
385 1 : CHK_PTR_NULL(endpointDesc);
386 1 : CHK_PTR_NULL(value);
387 1 : (void)HcommResMgrInit();
388 :
389 1 : return static_cast<HcommResult>(Endpoint::CheckFeature(*endpointDesc, featureType, *value));
390 : }
|
