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