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 : #include "endpoint.h"
11 : #include <functional>
12 : #include <chrono>
13 : #include <thread>
14 : #include "aicpu_ts_roce_endpoint.h"
15 : #include "cpu_roce_endpoint.h"
16 : #include "urma_endpoint.h"
17 : #include "ub_mem_endpoint.h"
18 : #include "uboe_endpoint.h"
19 : #include "ubg_endpoint.h"
20 : #include "cpu_urma_endpoint.h"
21 : #include "aicputs_hccs_endpoint.h"
22 : #include "hccp_nda.h"
23 : #include "adapter_rts_common.h"
24 : #include "rdma_handle_manager.h"
25 :
26 : namespace hcomm {
27 86 : static bool IsSupported(const EndpointDesc& endpointDesc)
28 : {
29 86 : bool protocolSupported = false;
30 86 : bool locTypeSupported = false;
31 86 : switch (endpointDesc.protocol) {
32 86 : case COMM_PROTOCOL_ROCE:
33 : case COMM_PROTOCOL_UBC_TP:
34 : case COMM_PROTOCOL_UBC_CTP:
35 : case COMM_PROTOCOL_UB_MEM:
36 : case COMM_PROTOCOL_PCIE:
37 : case COMM_PROTOCOL_UBOE:
38 : case COMM_PROTOCOL_UBG:
39 : case COMM_PROTOCOL_HCCS:
40 86 : protocolSupported = true;
41 86 : break;
42 0 : default:
43 0 : return false;
44 : }
45 86 : switch (endpointDesc.loc.locType) {
46 86 : case ENDPOINT_LOC_TYPE_DEVICE:
47 : case ENDPOINT_LOC_TYPE_HOST:
48 86 : locTypeSupported = true;
49 86 : break;
50 0 : default:
51 0 : return false;
52 : }
53 :
54 86 : return protocolSupported && locTypeSupported;
55 : }
56 :
57 317 : Endpoint::Endpoint(const EndpointDesc& endpointDesc) { endpointDesc_ = endpointDesc; }
58 :
59 4 : void Endpoint::DestroySharedJettyRaResources(SharedJettyCtx& ctx, Hccl::RdmaHandle rdmaHandle, bool ctxValid) const
60 : {
61 4 : if (ctx.handle != 0) {
62 0 : if (!ctxValid) {
63 0 : HCCL_WARNING("[Endpoint][%s] skip DestroyJetty, rdmaHandle=%p invalid.", __func__, ctx.rdmaHandle);
64 : } else {
65 0 : Hccl::HrtRaUbDestroyJetty(ctx.handle);
66 0 : HCCL_INFO(
67 : "[Endpoint][%s] destroyed shared jetty, handle[%llu]", __func__,
68 : static_cast<unsigned long long>(ctx.handle));
69 : }
70 : }
71 : // 销毁临时 connection 转移过来的 JFC（共享 jetty 模式下临时 connection 不自销毁 JFC）
72 4 : if (ctx.jfcHandle != 0 && ctx.rdmaHandle != nullptr) {
73 0 : if (!ctxValid) {
74 0 : HCCL_WARNING("[Endpoint][%s] skip DestroyJfc, rdmaHandle=%p invalid.", __func__, ctx.rdmaHandle);
75 : } else {
76 0 : Hccl::HrtRaUbDestroyJfc(rdmaHandle, ctx.jfcHandle);
77 0 : HCCL_INFO(
78 : "[Endpoint][%s] destroyed shared jfc, jfcHandle[%llu]", __func__,
79 : static_cast<unsigned long long>(ctx.jfcHandle));
80 : }
81 : }
82 4 : }
83 :
84 4 : void Endpoint::FreeSharedJettyPtrs(SharedJettyCtx& ctx) const
85 : {
86 4 : if (ctx.sqPiPtr != nullptr) {
87 4 : (void)hrtFree(ctx.sqPiPtr);
88 : }
89 4 : if (ctx.sqCiPtr != nullptr) {
90 4 : (void)hrtFree(ctx.sqCiPtr);
91 : }
92 4 : if (ctx.cqPiPtr != nullptr) {
93 4 : (void)hrtFree(ctx.cqPiPtr);
94 : }
95 4 : if (ctx.cqCiPtr != nullptr) {
96 4 : (void)hrtFree(ctx.cqCiPtr);
97 : }
98 4 : }
99 :
100 317 : Endpoint::~Endpoint()
101 : {
102 : // 防御性清理：若仍有共享 jetty 未释放（理论上 CheckEndpointDestroy 应已拦截）。
103 : // refCount == 0 时可安全强制销毁；refCount > 0 表示仍有 connection 持有 jetty 句柄，
104 : // 强制销毁会导致 use-after-free，此时仅告警不销毁（接受泄漏以避免更严重后果）。
105 317 : if (sharedJettyCtx_.valid && sharedJettyCtx_.handle != 0) {
106 0 : if (sharedJettyCtx_.refCount == 0) {
107 0 : HCCL_WARNING(
108 : "[Endpoint][~Endpoint] shared jetty still valid on destroy, handle[%llu], force destroy.",
109 : static_cast<unsigned long long>(sharedJettyCtx_.handle));
110 0 : RdmaHandle rdmaHandle = static_cast<Hccl::RdmaHandle>(sharedJettyCtx_.rdmaHandle);
111 : const bool ctxValid
112 0 : = rdmaHandle != nullptr && Hccl::RdmaHandleManager::GetInstance().IsHandleValid(rdmaHandle);
113 0 : if (!ctxValid) {
114 0 : HCCL_WARNING("[Endpoint][~Endpoint] skip shared jetty/jfc destroy, rdmaHandle=%p invalid.", rdmaHandle);
115 : } else {
116 0 : DestroySharedJettyRaResources(sharedJettyCtx_, rdmaHandle, ctxValid);
117 : }
118 0 : FreeSharedJettyPtrs(sharedJettyCtx_);
119 : } else {
120 0 : HCCL_WARNING(
121 : "[Endpoint][~Endpoint] shared jetty still in use, refCount[%u], handle[%llu], skip destroy "
122 : "to avoid use-after-free.",
123 : sharedJettyCtx_.refCount, static_cast<unsigned long long>(sharedJettyCtx_.handle));
124 : }
125 0 : sharedJettyCtx_ = SharedJettyCtx{};
126 : }
127 317 : }
128 :
129 : HcclResult
130 6 : Endpoint::AcquireSharedJetty(const std::function<HcclResult(SharedJettyCtx&)>& provideCtx, SharedJettyCtx& outCtx)
131 : {
132 : // 第一段（持锁）：检查是否已创建或正在创建。已创建则 refCount++ 返回；未创建则标记 creating。
133 : while (true) {
134 6 : std::unique_lock<std::mutex> lk(sharedJettyMtx_);
135 6 : if (sharedJettyCtx_.valid) {
136 1 : sharedJettyCtx_.refCount++;
137 1 : outCtx = sharedJettyCtx_;
138 1 : HCCL_INFO(
139 : "[Endpoint][AcquireSharedJetty] reuse shared jetty, handle[%llu], refCount[%u]",
140 : static_cast<unsigned long long>(outCtx.handle), sharedJettyCtx_.refCount);
141 1 : return HCCL_SUCCESS;
142 : }
143 5 : if (!sharedJettyCtx_.creating) {
144 : // 抢占创建权
145 5 : sharedJettyCtx_.creating = true;
146 5 : break;
147 : }
148 : // 其他线程正在创建：释放锁短暂 sleep 后重新检查，避免紧密 spin 占 CPU。
149 0 : lk.unlock();
150 0 : std::this_thread::sleep_for(std::chrono::milliseconds(2));
151 6 : }
152 :
153 : // 第二段（无锁）：执行首次创建回调（含网络建链 I/O，可能耗时数秒）。
154 : // 创建期间不持锁，其他线程的 Acquire 会在此循环等待，Release 不被阻塞。
155 5 : SharedJettyCtx createdCtx;
156 5 : HcclResult createRet = provideCtx(createdCtx);
157 5 : if (createRet != HCCL_SUCCESS) {
158 0 : std::lock_guard<std::mutex> lk(sharedJettyMtx_);
159 0 : sharedJettyCtx_.creating = false;
160 0 : HCCL_ERROR("[Endpoint][AcquireSharedJetty] provideCtx failed, ret[%d].", createRet);
161 0 : return createRet;
162 0 : }
163 :
164 : // 第三段（持锁）：写入缓存，清除 creating 标记，设置 refCount=1。
165 : {
166 5 : std::lock_guard<std::mutex> lk(sharedJettyMtx_);
167 5 : sharedJettyCtx_ = createdCtx;
168 5 : sharedJettyCtx_.valid = true;
169 5 : sharedJettyCtx_.creating = false;
170 5 : sharedJettyCtx_.refCount = 1;
171 5 : outCtx = sharedJettyCtx_;
172 5 : }
173 5 : HCCL_INFO(
174 : "[Endpoint][AcquireSharedJetty] created shared jetty, handle[%llu]",
175 : static_cast<unsigned long long>(outCtx.handle));
176 5 : return HCCL_SUCCESS;
177 : }
178 :
179 5 : HcclResult Endpoint::ReleaseSharedJetty()
180 : {
181 5 : std::lock_guard<std::mutex> lk(sharedJettyMtx_);
182 5 : if (!sharedJettyCtx_.valid) {
183 0 : HCCL_WARNING("[Endpoint][ReleaseSharedJetty] shared jetty already invalid, skip release.");
184 0 : return HCCL_SUCCESS;
185 : }
186 5 : if (sharedJettyCtx_.refCount == 0) {
187 0 : HCCL_WARNING("[Endpoint][ReleaseSharedJetty] refCount already 0, skip release.");
188 0 : return HCCL_SUCCESS;
189 : }
190 5 : sharedJettyCtx_.refCount--;
191 5 : HCCL_INFO(
192 : "[Endpoint][ReleaseSharedJetty] release shared jetty, handle[%llu], refCount[%u]",
193 : static_cast<unsigned long long>(sharedJettyCtx_.handle), sharedJettyCtx_.refCount);
194 5 : if (sharedJettyCtx_.refCount == 0) {
195 4 : const auto rdmaHandle = static_cast<Hccl::RdmaHandle>(sharedJettyCtx_.rdmaHandle);
196 4 : const bool ctxValid = rdmaHandle != nullptr && Hccl::RdmaHandleManager::GetInstance().IsHandleValid(rdmaHandle);
197 4 : DestroySharedJettyRaResources(sharedJettyCtx_, rdmaHandle, ctxValid);
198 4 : FreeSharedJettyPtrs(sharedJettyCtx_);
199 4 : sharedJettyCtx_ = SharedJettyCtx{};
200 : }
201 5 : return HCCL_SUCCESS;
202 5 : }
203 :
204 86 : HcclResult Endpoint::CreateEndpoint(const EndpointDesc& endpointDesc, std::unique_ptr<Endpoint>& endpointPtr)
205 : {
206 86 : if (!IsSupported(endpointDesc)) {
207 0 : HCCL_ERROR(
208 : "[%s]endpointDesc is not supported. endpointDesc.protocol [%d] endpointDesc.loc.locType [%d].", __func__,
209 : endpointDesc.protocol, endpointDesc.loc.locType);
210 0 : return HCCL_E_PARA;
211 : }
212 :
213 86 : HCCL_INFO(
214 : "[%s]endpointDesc.protocol [%d] endpointDesc.loc.locType [%d].", __func__, endpointDesc.protocol,
215 : endpointDesc.loc.locType);
216 :
217 86 : return CreateEndpointBase(endpointDesc, endpointPtr);
218 : }
219 :
220 86 : HcclResult Endpoint::CreateEndpointBase(const EndpointDesc& endpointDesc, std::unique_ptr<Endpoint>& endpointPtr)
221 : {
222 : using EndpointCreator = std::function<std::unique_ptr<Endpoint>(const EndpointDesc&)>;
223 : struct Entry {
224 : CommProtocol protocol;
225 : EndpointLocType locType;
226 : EndpointCreator creator;
227 : };
228 : static const Entry table[] = {
229 : {COMM_PROTOCOL_ROCE, ENDPOINT_LOC_TYPE_HOST,
230 22 : [](const EndpointDesc& d) {
231 18 : return std::make_unique<CpuRoceEndpoint>(d);
232 : }},
233 : {COMM_PROTOCOL_UBC_TP, ENDPOINT_LOC_TYPE_HOST,
234 4 : [](const EndpointDesc& d) {
235 0 : return std::make_unique<CpuUrmaEndpoint>(d);
236 : }},
237 : {COMM_PROTOCOL_UBC_CTP, ENDPOINT_LOC_TYPE_HOST,
238 4 : [](const EndpointDesc& d) {
239 0 : return std::make_unique<CpuUrmaEndpoint>(d);
240 : }},
241 : {COMM_PROTOCOL_UBC_TP, ENDPOINT_LOC_TYPE_DEVICE,
242 4 : [](const EndpointDesc& d) {
243 0 : return std::make_unique<UrmaEndpoint>(d);
244 : }},
245 : {COMM_PROTOCOL_UBC_CTP, ENDPOINT_LOC_TYPE_DEVICE,
246 22 : [](const EndpointDesc& d) {
247 18 : return std::make_unique<UrmaEndpoint>(d);
248 : }},
249 : {COMM_PROTOCOL_UB_MEM, ENDPOINT_LOC_TYPE_DEVICE,
250 21 : [](const EndpointDesc& d) {
251 17 : return std::make_unique<UbMemEndpoint>(d);
252 : }},
253 : {COMM_PROTOCOL_PCIE, ENDPOINT_LOC_TYPE_DEVICE,
254 4 : [](const EndpointDesc& d) {
255 0 : return std::make_unique<UbMemEndpoint>(d);
256 : }},
257 : {COMM_PROTOCOL_UBOE, ENDPOINT_LOC_TYPE_DEVICE,
258 17 : [](const EndpointDesc& d) {
259 13 : return std::make_unique<UboeEndpoint>(d);
260 : }},
261 : {COMM_PROTOCOL_UBG, ENDPOINT_LOC_TYPE_DEVICE,
262 6 : [](const EndpointDesc& d) {
263 2 : return std::make_unique<UbgEndpoint>(d);
264 : }},
265 : {COMM_PROTOCOL_ROCE, ENDPOINT_LOC_TYPE_DEVICE,
266 5 : [](const EndpointDesc& d) {
267 1 : return std::make_unique<AicpuTsRoceEndpoint>(d);
268 : }},
269 : {COMM_PROTOCOL_HCCS, ENDPOINT_LOC_TYPE_DEVICE,
270 4 : [](const EndpointDesc& d) {
271 15 : return std::make_unique<AicpuTsHccsEndpoint>(d);
272 : }},
273 86 : };
274 :
275 531 : for (const auto& entry : table) {
276 529 : if (entry.protocol == endpointDesc.protocol && entry.locType == endpointDesc.loc.locType) {
277 84 : EXCEPTION_CATCH(endpointPtr = entry.creator(endpointDesc), return HCCL_E_PTR);
278 84 : return HCCL_SUCCESS;
279 : }
280 : }
281 :
282 2 : HCCL_ERROR(
283 : "[%s] failed, endpointDesc.protocol [%d] and endpointDesc.loc.locType [%d] do not match.", __func__,
284 : endpointDesc.protocol, endpointDesc.loc.locType);
285 2 : return HCCL_E_PARA;
286 : }
287 :
288 1 : HcclResult Endpoint::CheckFeature(const EndpointDesc& endpointDesc, HcommEndpointFeatureType featureType, bool& value)
289 : {
290 1 : if (featureType == HCOMM_ENDPOINT_FEATURE_NDA) {
291 1 : if (endpointDesc.protocol != COMM_PROTOCOL_ROCE || endpointDesc.loc.locType != ENDPOINT_LOC_TYPE_HOST) {
292 0 : HCCL_WARNING(
293 : "[%s] not support NDA, protocol[%d], locType[%d]", __func__, endpointDesc.protocol,
294 : endpointDesc.loc.locType);
295 0 : value = false;
296 0 : return HCCL_SUCCESS;
297 : }
298 :
299 1 : Hccl::IpAddress ipAddr{};
300 1 : CHK_RET(CommAddrToIpAddress(endpointDesc.commAddr, ipAddr));
301 1 : s32 devId = 0;
302 1 : CHK_RET(hrtGetDevice(&devId));
303 1 : u32 devPhyId = 0;
304 1 : CHK_RET(hrtGetDevicePhyIdByIndex(devId, devPhyId));
305 :
306 1 : auto& rdmaHandleMgr = Hccl::RdmaHandleManager::GetInstance();
307 : void* rdmaHandle = static_cast<void*>(
308 1 : rdmaHandleMgr.GetByAddr(devPhyId, Hccl::LinkProtoType::RDMA, ipAddr, Hccl::PortDeploymentType::HOST_NET));
309 1 : CHK_PTR_NULL(rdmaHandle);
310 :
311 1 : s32 directFlag = 0;
312 1 : s32 ret = RaNdaGetDirectFlag(rdmaHandle, &directFlag);
313 1 : CHK_PRT_RET(
314 : ret != HCCL_SUCCESS, HCCL_ERROR("[%s] failed to get directFlag, ret[%d]", __func__, ret), HCCL_E_INTERNAL);
315 1 : value = (directFlag != DIRECT_FLAG_NOTSUPP);
316 1 : HCCL_INFO(
317 : "[%s] %s NDA, rdmaHandle[%p], directFlag[%d]", __func__, value ? "support" : "not support", rdmaHandle,
318 : directFlag);
319 : } else {
320 0 : HCCL_WARNING("[%s] unsupported featureType[%d]", __func__, featureType);
321 0 : value = false;
322 : }
323 :
324 1 : return HCCL_SUCCESS;
325 : }
326 : } // namespace hcomm
|
