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