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 94 : HcclResult ValidateEndpointDesc(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
100 : {
101 94 : CHK_PTR_NULL(endpoint);
102 93 : CHK_PTR_NULL(endpointHandle);
103 92 : 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 92 : 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 : auto* ep = GetEndpointMap().GetEndpoint(handle);
125 19 : CHK_PRT_RET(ep == nullptr, HCCL_ERROR("[%s] endpoint not found, handle[%p]", __func__, handle), HCCL_E_NOT_FOUND);
126 19 : ep->AttachMonitor(devLogicIdSigned);
127 19 : CHK_RET(ep->RegisterToEndpointMonitor(devLogicIdSigned, handle));
128 19 : return HCCL_SUCCESS;
129 : }
130 :
131 4 : HcommResult CreatePluginEndpointHolder(
132 : const EndpointDesc* endpoint, const NicPluginEntry* pluginEntry, EndpointHandle* endpointHandle)
133 : {
134 4 : CHK_PTR_NULL(endpoint);
135 4 : CHK_PTR_NULL(pluginEntry);
136 4 : CHK_PTR_NULL(endpointHandle);
137 4 : void* pluginCtx = nullptr;
138 4 : HcommNicEndpointOps* pluginOps = nullptr;
139 4 : HcommResult ret = HCCL_SUCCESS;
140 4 : ret = static_cast<HcommResult>(pluginEntry->createEndpoint(endpoint, &pluginCtx, &pluginOps));
141 4 : CHK_PRT_RET(
142 : ret != HCCL_SUCCESS,
143 : HCCL_ERROR(
144 : "[NicPlugin][%s] plugin createEndpoint failed, ret[%d], protocol[%d].", __func__, ret, endpoint->protocol),
145 : ret);
146 :
147 4 : CHK_PRT_RET(
148 : !ValidateEndpointOps(pluginOps),
149 : HCCL_ERROR(
150 : "[NicPlugin][%s] plugin endpoint ops validation failed, protocol[%d].", __func__, endpoint->protocol),
151 : HCCL_E_PARA);
152 :
153 : // 确保HcommNicEndpointOps各接口非空实现，后续调用处无需校验
154 4 : HcommNicEndpointOps* pluginHolderOps = nullptr;
155 4 : ret = FillDefaultEndpointOps(pluginOps, &pluginHolderOps);
156 4 : CHK_PRT_RET(
157 : ret != HCCL_SUCCESS,
158 : HCCL_ERROR(
159 : "[NicPlugin][%s] FillDefaultEndpointOps failed, ret[%d], protocol[%d].", __func__, ret, endpoint->protocol),
160 : ret);
161 :
162 4 : ret = static_cast<HcommResult>(pluginHolderOps->init(pluginCtx));
163 4 : if (ret != HCCL_SUCCESS) {
164 0 : int32_t destroyRet = pluginHolderOps->destroy(pluginCtx);
165 0 : if (destroyRet != HCCL_SUCCESS) {
166 0 : HCCL_WARNING("[%s] plugin endpoint destroy failed after init failure, ret[%d].", __func__, destroyRet);
167 : }
168 0 : delete pluginHolderOps;
169 0 : HCCL_ERROR("[NicPlugin][%s] plugin endpoint init failed, ret[%d].", __func__, ret);
170 0 : return ret;
171 : }
172 :
173 4 : auto holder = std::make_unique<PluginEndpointHolder>(*endpoint, pluginEntry);
174 4 : holder->SetNicEndpointCtx(pluginHolderOps, pluginCtx);
175 4 : const EndpointHandle handle = reinterpret_cast<EndpointHandle>(holder.get());
176 4 : EXCEPTION_CATCH(GetEndpointMap().AddEndpoint(handle, std::move(holder)), return HCCL_E_INTERNAL);
177 4 : *endpointHandle = handle;
178 4 : HCCL_INFO(
179 : "[NicPlugin][%s] plugin endpoint created, protocol[%d], handle[%p].", __func__, endpoint->protocol, handle);
180 4 : return HCCL_SUCCESS;
181 4 : }
182 :
183 88 : HcclResult CreateBuiltinEndpoint(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
184 : {
185 88 : CHK_RET(RefreshEndpointContext(*endpoint));
186 88 : std::unique_ptr<Endpoint> endpointPtr = nullptr;
187 88 : HcclResult ret = Endpoint::CreateEndpoint(*endpoint, endpointPtr);
188 88 : if (ret != HCCL_SUCCESS) {
189 2 : HCCL_ERROR("call Endpoint::CreateEndpoint failed");
190 2 : return ret;
191 : }
192 86 : CHK_PTR_NULL(endpointPtr);
193 86 : ret = endpointPtr->Init();
194 86 : if (ret != HCCL_SUCCESS) {
195 6 : HCCL_ERROR("call endpointPtr->Init failed");
196 6 : return ret;
197 : }
198 :
199 80 : endpointPtr->SetNicEndpointCtx(&g_BuiltinEndpointOps, endpointPtr.get());
200 :
201 80 : const EndpointHandle handle = reinterpret_cast<EndpointHandle>(endpointPtr.get());
202 80 : CHK_PTR_NULL(handle);
203 80 : EXCEPTION_CATCH(GetEndpointMap().AddEndpoint(handle, std::move(endpointPtr)), return HCCL_E_INTERNAL);
204 80 : *endpointHandle = handle;
205 80 : CHK_RET(RegisterDeviceEndpointMonitorIfNeeded(endpoint, handle));
206 80 : HCCL_INFO(
207 : "[%s] endpointDesc.protocol [%d] and endpointDesc.loc.locType [%d] create endpointHandle [%p] done.", __func__,
208 : endpoint->protocol, endpoint->loc.locType, handle);
209 80 : return HCCL_SUCCESS;
210 88 : }
211 : } // namespace
212 :
213 5 : HcommResult HcommEndpointGetDescNum(int32_t deviceLogicId, uint32_t* descNum)
214 : {
215 : EXCEPTION_HANDLE_BEGIN
216 9 : CHK_PTR_NULL(descNum);
217 5 : *descNum = 0;
218 5 : HcommResMgr::RegisterDeviceResetCallback();
219 5 : HcommResMgr::RegisterDeviceRefreshCallback();
220 :
221 5 : CHK_PRT_RET(
222 : deviceLogicId < 0, HCCL_ERROR("[%s] deviceLogicId[%d] is invalid.", __func__, deviceLogicId), HCCL_E_PARA);
223 5 : DevType deviceType = DevType::DEV_TYPE_COUNT;
224 5 : CHK_RET(hrtGetDeviceType(deviceType));
225 5 : CHK_PRT_RET(
226 : deviceType != DevType::DEV_TYPE_950,
227 : HCCL_ERROR(
228 : "[%s] endpoint query only supports DEV_TYPE_950, current deviceType[%d].", __func__,
229 : static_cast<int32_t>(deviceType)),
230 : HCCL_E_NOT_SUPPORT);
231 3 : uint32_t devicePhyId = 0;
232 3 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId));
233 3 : std::vector<DevEidInfo> eidInfos;
234 3 : CHK_RET(GetDeviceEidInfos(deviceLogicId, devicePhyId, eidInfos));
235 :
236 1 : *descNum = static_cast<uint32_t>(eidInfos.size());
237 1 : HCCL_INFO(
238 : "[%s] success, deviceLogicId[%d], devicePhyId[%u], descNum[%u].", __func__, deviceLogicId, devicePhyId,
239 : *descNum);
240 3 : EXCEPTION_HANDLE_END
241 1 : return HCCL_SUCCESS;
242 : }
243 :
244 2 : HcommResult HcommEndpointGetDescs(int32_t deviceLogicId, uint32_t* descNum, EndpointDesc* endpointDescs)
245 : {
246 : EXCEPTION_HANDLE_BEGIN
247 2 : CHK_PTR_NULL(descNum);
248 2 : CHK_PTR_NULL(endpointDescs);
249 2 : HcommResMgr::RegisterDeviceResetCallback();
250 2 : HcommResMgr::RegisterDeviceRefreshCallback();
251 :
252 2 : uint32_t devicePhyId = 0;
253 2 : std::vector<DevEidInfo> eidInfos;
254 2 : CHK_PRT_RET(
255 : deviceLogicId < 0, HCCL_ERROR("[%s] deviceLogicId[%d] is invalid.", __func__, deviceLogicId), HCCL_E_PARA);
256 2 : const uint32_t capacity = *descNum;
257 2 : DevType deviceType = DevType::DEV_TYPE_COUNT;
258 2 : CHK_RET(hrtGetDeviceType(deviceType));
259 2 : CHK_PRT_RET(
260 : deviceType != DevType::DEV_TYPE_950,
261 : HCCL_ERROR(
262 : "[%s] endpoint query only supports DEV_TYPE_950, current deviceType[%d].", __func__,
263 : static_cast<int32_t>(deviceType)),
264 : HCCL_E_NOT_SUPPORT);
265 2 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId));
266 2 : CHK_RET(GetDeviceEidInfos(deviceLogicId, devicePhyId, eidInfos));
267 2 : CHK_PRT_RET(
268 : eidInfos.size() > capacity,
269 : HCCL_ERROR(
270 : "[%s] endpointDescs capacity[%u] is smaller than required count[%zu].", __func__, capacity,
271 : eidInfos.size()),
272 : HCCL_E_PARA);
273 :
274 2 : const uint32_t actualNum = static_cast<uint32_t>(eidInfos.size());
275 2 : const HcommResult initRet = EndpointDescInit(endpointDescs, actualNum);
276 2 : CHK_PRT_RET(
277 : initRet != HCOMM_SUCCESS, HCCL_ERROR("[%s] EndpointDescInit failed, ret[%d].", __func__, initRet), initRet);
278 5 : for (uint32_t i = 0; i < actualNum; ++i) {
279 3 : CHK_RET(FillEndpointDesc(devicePhyId, eidInfos[i], endpointDescs[i]));
280 : }
281 :
282 2 : *descNum = actualNum;
283 2 : HCCL_INFO(
284 : "[%s] success, deviceLogicId[%d], devicePhyId[%u], descNum[%u].", __func__, deviceLogicId, devicePhyId,
285 : *descNum);
286 2 : EXCEPTION_HANDLE_END
287 2 : return HCCL_SUCCESS;
288 : }
289 :
290 23 : HcommResult HcommEndpointGet(EndpointHandle endpointHandle, void** endpoint) // 根据endpointHandle返回Endpoint对象指针
291 : {
292 23 : CHK_PTR_NULL(endpoint);
293 :
294 20 : auto it = GetEndpointMap().GetEndpoint(endpointHandle);
295 20 : CHK_PRT_RET(
296 : it == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
297 : HCCL_E_NOT_FOUND);
298 :
299 18 : *endpoint = static_cast<void*>(it);
300 18 : HCCL_INFO(
301 : "[%s] START. endpointHandle[%p] endpoint[%p].", __func__, static_cast<void*>(endpointHandle),
302 : static_cast<void*>(endpoint));
303 18 : return HCCL_SUCCESS;
304 : }
305 :
306 94 : HcommResult HcommEndpointCreate(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
307 : {
308 : EXCEPTION_HANDLE_BEGIN
309 94 : CHK_RET(ValidateEndpointDesc(endpoint, endpointHandle));
310 92 : if (endpoint->loc.locType == ENDPOINT_LOC_TYPE_HOST) {
311 24 : const NicPluginEntry* pluginEntry = FindHostNicPlugin(endpoint->protocol);
312 24 : if (pluginEntry != nullptr) {
313 4 : return CreatePluginEndpointHolder(endpoint, pluginEntry, endpointHandle);
314 : }
315 : }
316 88 : (void)HcommResMgrInit();
317 88 : CHK_RET(CreateBuiltinEndpoint(endpoint, endpointHandle));
318 80 : HcommResMgr::RegisterDeviceResetCallback();
319 80 : HcommResMgr::RegisterDeviceRefreshCallback();
320 0 : EXCEPTION_HANDLE_END
321 80 : return HCCL_SUCCESS;
322 : }
323 :
324 55 : HcommResult HcommEndpointDestroy(EndpointHandle endpointHandle)
325 : {
326 55 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].", __func__, endpointHandle);
327 55 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
328 55 : if (endpoint != nullptr && endpoint->GetNicOps() != nullptr && endpoint->GetNicOps() != &g_BuiltinEndpointOps) {
329 4 : HCCL_INFO("[NicPlugin][%s] destroy plugin endpoint.", __func__);
330 : // 先摘除 SharedJettyMgr 中该 plugin endpoint 的 channel 注册记录, 避免 plugin 句柄复用误判
331 4 : hcomm::SharedJettyMgr::GetInstance().UnregisterEndpoint(endpointHandle);
332 4 : endpoint->ReleaseEndpointMonitor(endpointHandle);
333 4 : auto ret = GetEndpointMap().RemoveEndpoint(endpointHandle);
334 4 : CHK_PRT_RET(
335 : ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]", __func__, endpointHandle),
336 : HCCL_E_NOT_FOUND);
337 4 : return HCCL_SUCCESS;
338 : }
339 51 : (void)HcommResMgrInit();
340 : // 需校验共享 jetty channel 是否已全部销毁，否则残留 channel 持有的 jetty 引用会在 endpoint 销毁后成为悬空引用。
341 51 : HcclResult jettyRet = hcomm::SharedJettyMgr::GetInstance().CheckEndpointDestroy(endpointHandle);
342 51 : CHK_PRT_RET(
343 : jettyRet != HCCL_SUCCESS,
344 : HCCL_ERROR(
345 : "[%s] cannot destroy endpointHandle[0x%llx], shared jetty channels still exist.", __func__, endpointHandle),
346 : jettyRet);
347 51 : if (endpoint != nullptr) {
348 49 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
349 49 : endpoint->ReleaseEndpointMonitor(endpointHandle);
350 : }
351 : // 在 RemoveEndpoint 释放 Endpoint 对象前摘除 SharedJettyMgr 反查记录，
352 : // 避免 Endpoint* 复用误判；此处单例确定存活（运行期），不依赖 ~Endpoint 调用以规避静态析构顺序风险。
353 51 : hcomm::SharedJettyMgr::GetInstance().UnregisterEndpoint(endpointHandle);
354 51 : auto ret = GetEndpointMap().RemoveEndpoint(endpointHandle);
355 51 : CHK_PRT_RET(
356 : ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]", __func__, endpointHandle),
357 : HCCL_E_NOT_FOUND);
358 49 : return HCCL_SUCCESS;
359 : }
360 :
361 25 : HcommResult HcommEndpointStartListen(EndpointHandle endpointHandle, uint32_t port, HcommEndpointListenConfig* config)
362 : {
363 : (void)config;
364 25 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
365 25 : CHK_PRT_RET(
366 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
367 : HCCL_E_NOT_FOUND);
368 24 : CHK_RET(endpoint->ServerSocketListen(port));
369 24 : return HCCL_SUCCESS;
370 : }
371 :
372 3 : HcommResult HcommEndpointStopListen(EndpointHandle endpointHandle, uint32_t port)
373 : {
374 3 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
375 3 : CHK_PRT_RET(
376 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
377 : HCCL_E_NOT_FOUND);
378 2 : CHK_RET(endpoint->ServerSocketStopListen(port));
379 2 : return HCCL_SUCCESS;
380 : }
381 :
382 8 : HcommResult HcommEndpointGetListenPort(EndpointHandle endpointHandle, uint32_t* port)
383 : {
384 8 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
385 8 : CHK_PRT_RET(
386 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
387 : HCCL_E_NOT_FOUND);
388 5 : return static_cast<HcclResult>(endpoint->GetNicOps()->getListenPort(endpoint->GetNicCtx(), port));
389 : }
390 :
391 : HcommResult
392 1 : HcommEndpointCheckFeature(HcommEndpointFeatureType featureType, const EndpointDesc* endpointDesc, bool* value)
393 : {
394 1 : CHK_PTR_NULL(endpointDesc);
395 1 : CHK_PTR_NULL(value);
396 1 : (void)HcommResMgrInit();
397 :
398 1 : return static_cast<HcommResult>(Endpoint::CheckFeature(*endpointDesc, featureType, *value));
399 : }
|
