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 :
220 5 : CHK_PRT_RET(
221 : deviceLogicId < 0, HCCL_ERROR("[%s] deviceLogicId[%d] is invalid.", __func__, deviceLogicId), HCCL_E_PARA);
222 5 : DevType deviceType = DevType::DEV_TYPE_COUNT;
223 5 : CHK_RET(hrtGetDeviceType(deviceType));
224 5 : CHK_PRT_RET(
225 : deviceType != DevType::DEV_TYPE_950,
226 : HCCL_ERROR(
227 : "[%s] endpoint query only supports DEV_TYPE_950, current deviceType[%d].", __func__,
228 : static_cast<int32_t>(deviceType)),
229 : HCCL_E_NOT_SUPPORT);
230 3 : uint32_t devicePhyId = 0;
231 3 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId));
232 3 : std::vector<DevEidInfo> eidInfos;
233 3 : CHK_RET(GetDeviceEidInfos(deviceLogicId, devicePhyId, eidInfos));
234 :
235 1 : *descNum = static_cast<uint32_t>(eidInfos.size());
236 1 : HCCL_INFO(
237 : "[%s] success, deviceLogicId[%d], devicePhyId[%u], descNum[%u].", __func__, deviceLogicId, devicePhyId,
238 : *descNum);
239 3 : EXCEPTION_HANDLE_END
240 1 : return HCCL_SUCCESS;
241 : }
242 :
243 2 : HcommResult HcommEndpointGetDescs(int32_t deviceLogicId, uint32_t* descNum, EndpointDesc* endpointDescs)
244 : {
245 : EXCEPTION_HANDLE_BEGIN
246 2 : CHK_PTR_NULL(descNum);
247 2 : CHK_PTR_NULL(endpointDescs);
248 2 : HcommResMgr::RegisterDeviceResetCallback();
249 :
250 2 : uint32_t devicePhyId = 0;
251 2 : std::vector<DevEidInfo> eidInfos;
252 2 : CHK_PRT_RET(
253 : deviceLogicId < 0, HCCL_ERROR("[%s] deviceLogicId[%d] is invalid.", __func__, deviceLogicId), HCCL_E_PARA);
254 2 : const uint32_t capacity = *descNum;
255 2 : DevType deviceType = DevType::DEV_TYPE_COUNT;
256 2 : CHK_RET(hrtGetDeviceType(deviceType));
257 2 : CHK_PRT_RET(
258 : deviceType != DevType::DEV_TYPE_950,
259 : HCCL_ERROR(
260 : "[%s] endpoint query only supports DEV_TYPE_950, current deviceType[%d].", __func__,
261 : static_cast<int32_t>(deviceType)),
262 : HCCL_E_NOT_SUPPORT);
263 2 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(deviceLogicId), devicePhyId));
264 2 : CHK_RET(GetDeviceEidInfos(deviceLogicId, devicePhyId, eidInfos));
265 2 : CHK_PRT_RET(
266 : eidInfos.size() > capacity,
267 : HCCL_ERROR(
268 : "[%s] endpointDescs capacity[%u] is smaller than required count[%zu].", __func__, capacity,
269 : eidInfos.size()),
270 : HCCL_E_PARA);
271 :
272 2 : const uint32_t actualNum = static_cast<uint32_t>(eidInfos.size());
273 2 : const HcommResult initRet = EndpointDescInit(endpointDescs, actualNum);
274 2 : CHK_PRT_RET(
275 : initRet != HCOMM_SUCCESS, HCCL_ERROR("[%s] EndpointDescInit failed, ret[%d].", __func__, initRet), initRet);
276 5 : for (uint32_t i = 0; i < actualNum; ++i) {
277 3 : CHK_RET(FillEndpointDesc(devicePhyId, eidInfos[i], endpointDescs[i]));
278 : }
279 :
280 2 : *descNum = actualNum;
281 2 : HCCL_INFO(
282 : "[%s] success, deviceLogicId[%d], devicePhyId[%u], descNum[%u].", __func__, deviceLogicId, devicePhyId,
283 : *descNum);
284 2 : EXCEPTION_HANDLE_END
285 2 : return HCCL_SUCCESS;
286 : }
287 :
288 23 : HcommResult HcommEndpointGet(EndpointHandle endpointHandle, void** endpoint) // 根据endpointHandle返回Endpoint对象指针
289 : {
290 23 : CHK_PTR_NULL(endpoint);
291 :
292 20 : auto it = GetEndpointMap().GetEndpoint(endpointHandle);
293 20 : CHK_PRT_RET(
294 : it == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
295 : HCCL_E_NOT_FOUND);
296 :
297 18 : *endpoint = static_cast<void*>(it);
298 18 : HCCL_INFO(
299 : "[%s] START. endpointHandle[%p] endpoint[%p].", __func__, static_cast<void*>(endpointHandle),
300 : static_cast<void*>(endpoint));
301 18 : return HCCL_SUCCESS;
302 : }
303 :
304 94 : HcommResult HcommEndpointCreate(const EndpointDesc* endpoint, EndpointHandle* endpointHandle)
305 : {
306 : EXCEPTION_HANDLE_BEGIN
307 94 : CHK_RET(ValidateEndpointDesc(endpoint, endpointHandle));
308 92 : if (endpoint->loc.locType == ENDPOINT_LOC_TYPE_HOST) {
309 24 : const NicPluginEntry* pluginEntry = FindHostNicPlugin(endpoint->protocol);
310 24 : if (pluginEntry != nullptr) {
311 4 : return CreatePluginEndpointHolder(endpoint, pluginEntry, endpointHandle);
312 : }
313 : }
314 88 : (void)HcommResMgrInit();
315 88 : CHK_RET(CreateBuiltinEndpoint(endpoint, endpointHandle));
316 80 : HcommResMgr::RegisterDeviceResetCallback();
317 0 : EXCEPTION_HANDLE_END
318 80 : return HCCL_SUCCESS;
319 : }
320 :
321 55 : HcommResult HcommEndpointDestroy(EndpointHandle endpointHandle)
322 : {
323 55 : HCCL_INFO("[%s] START. endpointHandle[0x%llx].", __func__, endpointHandle);
324 55 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
325 55 : if (endpoint != nullptr && endpoint->GetNicOps() != nullptr && endpoint->GetNicOps() != &g_BuiltinEndpointOps) {
326 4 : HCCL_INFO("[NicPlugin][%s] destroy plugin endpoint.", __func__);
327 : // 先摘除 SharedJettyMgr 中该 plugin endpoint 的 channel 注册记录, 避免 plugin 句柄复用误判
328 4 : hcomm::SharedJettyMgr::GetInstance().UnregisterEndpoint(endpointHandle);
329 4 : endpoint->ReleaseEndpointMonitor(endpointHandle);
330 4 : auto ret = GetEndpointMap().RemoveEndpoint(endpointHandle);
331 4 : CHK_PRT_RET(
332 : ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]", __func__, endpointHandle),
333 : HCCL_E_NOT_FOUND);
334 4 : return HCCL_SUCCESS;
335 : }
336 51 : (void)HcommResMgrInit();
337 : // 需校验共享 jetty channel 是否已全部销毁，否则残留 channel 持有的 jetty 引用会在 endpoint 销毁后成为悬空引用。
338 51 : HcclResult jettyRet = hcomm::SharedJettyMgr::GetInstance().CheckEndpointDestroy(endpointHandle);
339 51 : CHK_PRT_RET(
340 : jettyRet != HCCL_SUCCESS,
341 : HCCL_ERROR(
342 : "[%s] cannot destroy endpointHandle[0x%llx], shared jetty channels still exist.", __func__, endpointHandle),
343 : jettyRet);
344 51 : if (endpoint != nullptr) {
345 49 : CHK_RET(RefreshEndpointContext(endpoint->GetEndpointDesc()));
346 49 : endpoint->ReleaseEndpointMonitor(endpointHandle);
347 : }
348 : // 在 RemoveEndpoint 释放 Endpoint 对象前摘除 SharedJettyMgr 反查记录，
349 : // 避免 Endpoint* 复用误判；此处单例确定存活（运行期），不依赖 ~Endpoint 调用以规避静态析构顺序风险。
350 51 : hcomm::SharedJettyMgr::GetInstance().UnregisterEndpoint(endpointHandle);
351 51 : auto ret = GetEndpointMap().RemoveEndpoint(endpointHandle);
352 51 : CHK_PRT_RET(
353 : ret == false, HCCL_ERROR("[%s] endpoint not found, endpointHandle[0x%llx]", __func__, endpointHandle),
354 : HCCL_E_NOT_FOUND);
355 49 : return HCCL_SUCCESS;
356 : }
357 :
358 25 : HcommResult HcommEndpointStartListen(EndpointHandle endpointHandle, uint32_t port, HcommEndpointListenConfig* config)
359 : {
360 : (void)config;
361 25 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
362 25 : CHK_PRT_RET(
363 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
364 : HCCL_E_NOT_FOUND);
365 24 : CHK_RET(endpoint->ServerSocketListen(port));
366 24 : return HCCL_SUCCESS;
367 : }
368 :
369 3 : HcommResult HcommEndpointStopListen(EndpointHandle endpointHandle, uint32_t port)
370 : {
371 3 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
372 3 : CHK_PRT_RET(
373 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
374 : HCCL_E_NOT_FOUND);
375 2 : CHK_RET(endpoint->ServerSocketStopListen(port));
376 2 : return HCCL_SUCCESS;
377 : }
378 :
379 8 : HcommResult HcommEndpointGetListenPort(EndpointHandle endpointHandle, uint32_t* port)
380 : {
381 8 : auto endpoint = GetEndpointMap().GetEndpoint(endpointHandle);
382 8 : CHK_PRT_RET(
383 : endpoint == nullptr, HCCL_ERROR("[%s] endpoint not found, endpointHandle[%p]", __func__, endpointHandle),
384 : HCCL_E_NOT_FOUND);
385 5 : return static_cast<HcclResult>(endpoint->GetNicOps()->getListenPort(endpoint->GetNicCtx(), port));
386 : }
387 :
388 : HcommResult
389 1 : HcommEndpointCheckFeature(HcommEndpointFeatureType featureType, const EndpointDesc* endpointDesc, bool* value)
390 : {
391 1 : CHK_PTR_NULL(endpointDesc);
392 1 : CHK_PTR_NULL(value);
393 1 : (void)HcommResMgrInit();
394 :
395 1 : return static_cast<HcommResult>(Endpoint::CheckFeature(*endpointDesc, featureType, *value));
396 : }
|
