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 "jetty_context.h"
12 : #include "log.h"
13 : #include "adapter_rts_common.h"
14 :
15 : namespace hcomm {
16 : namespace {
17 :
18 4 : bool IsSameRemoteJetty(const JettyContext::SharedRemoteJettyCtx& ctx, const uint8_t* remoteQpKey, uint32_t keySize)
19 : {
20 4 : return ctx.remoteQpKey.size() == keySize && std::memcmp(ctx.remoteQpKey.data(), remoteQpKey, keySize) == 0;
21 : }
22 : } // namespace
23 :
24 52 : JettyContext::~JettyContext()
25 : {
26 26 : std::lock_guard<std::mutex> lk(mtx_);
27 26 : if (inner_.valid && inner_.handle != 0) {
28 9 : if (inner_.refCount == 0) {
29 : // 与 DevUbConnection::ReleaseResource 的 ctxValid 防御一致：进程退出/DeInit 阶段
30 : // RdmaHandleManager 可能已销毁，句柄可能已失效，用失效句柄 unimport/destroy 会崩溃。
31 0 : const bool rdmaValid = (inner_.rdmaHandle != nullptr)
32 0 : && Hccl::RdmaHandleManager::GetInstance().IsHandleValid(
33 0 : static_cast<Hccl::RdmaHandle>(inner_.rdmaHandle));
34 0 : if (!rdmaValid) {
35 0 : HCCL_WARNING(
36 : "[JettyContext][~JettyContext] shared jetty still valid on destroy but rdmaHandle "
37 : "invalid, skip unimport/destroy to avoid using invalid handle, handle[%llu].",
38 : static_cast<unsigned long long>(inner_.handle));
39 0 : return;
40 : }
41 0 : HCCL_WARNING(
42 : "[JettyContext][~JettyContext] shared jetty still valid on destroy, handle[%llu], "
43 : "force destroy.",
44 : static_cast<unsigned long long>(inner_.handle));
45 0 : UnimportSharedRemoteJettys(inner_);
46 0 : DestroyJettyResources(inner_);
47 : } else {
48 9 : HCCL_WARNING(
49 : "[JettyContext][~JettyContext] shared jetty still in use, refCount[%u], handle[%llu], "
50 : "skip destroy to avoid use-after-free.",
51 : inner_.refCount, static_cast<unsigned long long>(inner_.handle));
52 : }
53 : }
54 52 : }
55 :
56 27 : HcclResult JettyContext::Acquire(const std::function<HcclResult(Ctx&)>& provideCtx, Ctx& outCtx)
57 : {
58 : // 第一段（持锁）：检查是否已创建或正在创建。已创建则 refCount++ 返回；未创建则标记 creating。
59 : // 超时上限 32s（覆盖 provideCtx 内部 16s jetty 创建 + 余量），避免创建线程异常崩溃后其他线程永久阻塞。
60 : {
61 27 : std::unique_lock<std::mutex> lk(mtx_);
62 27 : if (!cv_.wait_for(lk, std::chrono::seconds(32), [this] {
63 28 : return inner_.valid || !inner_.creating;
64 : })) {
65 0 : HCCL_ERROR("[JettyContext][Acquire] wait for shared jetty creation timeout[32s].");
66 0 : return HCCL_E_TIMEOUT;
67 : }
68 27 : if (inner_.valid) {
69 10 : inner_.refCount++;
70 10 : outCtx = InnerToCtx(inner_);
71 10 : HCCL_INFO(
72 : "[JettyContext][Acquire] reuse shared jetty, handle[%llu], refCount[%u]",
73 : static_cast<unsigned long long>(outCtx.handle), inner_.refCount);
74 10 : return HCCL_SUCCESS;
75 : }
76 : // 抢占创建权
77 17 : inner_.creating = true;
78 27 : }
79 :
80 : // 第二段（无锁）：执行首次创建回调（含网络建链 I/O，可能耗时数秒）。
81 : // 创建期间不持锁，其他线程的 Acquire 在 cv_ 上等待，Release 不被阻塞。
82 17 : Ctx createdCtx;
83 17 : HcclResult createRet = provideCtx(createdCtx);
84 16 : if (createRet != HCCL_SUCCESS) {
85 1 : std::lock_guard<std::mutex> lk(mtx_);
86 1 : inner_.creating = false;
87 1 : cv_.notify_all();
88 1 : HCCL_ERROR("[JettyContext][Acquire] provideCtx failed, ret[%d].", createRet);
89 1 : return createRet;
90 1 : }
91 :
92 : // 第三段（持锁）：写入缓存，清除 creating 标记，设置 refCount=1。
93 : {
94 15 : std::lock_guard<std::mutex> lk(mtx_);
95 15 : inner_.handle = createdCtx.handle;
96 15 : inner_.handlePtr = createdCtx.handlePtr;
97 15 : inner_.jettyId = createdCtx.jettyId;
98 15 : inner_.sqBuffVa = createdCtx.sqBuffVa;
99 15 : inner_.dbAddr = createdCtx.dbAddr;
100 15 : inner_.keySize = createdCtx.keySize;
101 15 : inner_.sqDepth = createdCtx.sqDepth;
102 15 : inner_.sqPiPtr = createdCtx.sqPiPtr;
103 15 : inner_.sqCiPtr = createdCtx.sqCiPtr;
104 15 : inner_.cqPiPtr = createdCtx.cqPiPtr;
105 15 : inner_.cqCiPtr = createdCtx.cqCiPtr;
106 15 : inner_.queueIndexMemSize = createdCtx.queueIndexMemSize;
107 15 : inner_.rdmaHandle = createdCtx.rdmaHandle;
108 15 : inner_.jfcHandle = createdCtx.jfcHandle;
109 15 : inner_.cqInfo = createdCtx.cqInfo;
110 15 : inner_.localPsn = createdCtx.localPsn;
111 15 : if (createdCtx.keySize > 0 && createdCtx.keySize <= Hccl::HRT_UB_QP_KEY_MAX_LEN) {
112 : errno_t cpyRet
113 10 : = memcpy_s(inner_.localQpKey, Hccl::HRT_UB_QP_KEY_MAX_LEN, createdCtx.localQpKey, createdCtx.keySize);
114 10 : if (cpyRet != EOK) {
115 : // memcpy_s 失败：inner_ 已写入部分字段（handle/PI·CI 指针等）但 valid 仍为 false，
116 : // 需销毁已写入的 device 资源避免泄漏，再清空 inner_ 让其他线程可重新创建。
117 0 : DestroyJettyResources(inner_);
118 0 : inner_ = Inner{};
119 0 : inner_.creating = false;
120 0 : cv_.notify_all();
121 0 : HCCL_ERROR("[JettyContext][Acquire] memcpy_s localQpKey failed, ret[%d].", cpyRet);
122 0 : return HCCL_E_INTERNAL;
123 : }
124 : }
125 15 : inner_.valid = true;
126 15 : inner_.creating = false;
127 15 : inner_.refCount = 1;
128 15 : outCtx = InnerToCtx(inner_);
129 15 : cv_.notify_all();
130 15 : }
131 15 : HCCL_INFO(
132 : "[JettyContext][Acquire] created shared jetty, handle[%llu]", static_cast<unsigned long long>(outCtx.handle));
133 15 : return HCCL_SUCCESS;
134 : }
135 :
136 11 : HcclResult JettyContext::Release()
137 : {
138 11 : std::lock_guard<std::mutex> lk(mtx_);
139 11 : if (!inner_.valid) {
140 3 : HCCL_WARNING("[JettyContext][Release] shared jetty already invalid, skip release.");
141 3 : return HCCL_SUCCESS;
142 : }
143 8 : if (inner_.refCount == 0) {
144 1 : HCCL_WARNING("[JettyContext][Release] refCount already 0, skip release.");
145 1 : return HCCL_SUCCESS;
146 : }
147 7 : inner_.refCount--;
148 7 : HCCL_INFO(
149 : "[JettyContext][Release] release shared jetty, handle[%llu], refCount[%u]",
150 : static_cast<unsigned long long>(inner_.handle), inner_.refCount);
151 7 : if (inner_.refCount == 0) {
152 5 : UnimportSharedRemoteJettys(inner_);
153 5 : DestroyJettyResources(inner_);
154 5 : inner_ = Inner{};
155 : }
156 7 : return HCCL_SUCCESS;
157 11 : }
158 :
159 9 : HcclResult JettyContext::AcquireSharedRemoteJetty(
160 : const uint8_t* remoteQpKey, uint32_t keySize, bool& needImport, uint64_t& handle, void*& handlePtr, uint32_t& tpn)
161 : {
162 9 : CHK_PTR_NULL(remoteQpKey);
163 8 : CHK_PRT_RET(
164 : keySize == 0 || keySize > Hccl::HRT_UB_QP_KEY_MAX_LEN,
165 : HCCL_ERROR("[%s] invalid keySize[%u].", __func__, keySize), HCCL_E_PARA);
166 :
167 7 : std::lock_guard<std::mutex> lk(mtx_);
168 7 : CHK_PRT_RET(!inner_.valid, HCCL_ERROR("[%s] shared local jetty is invalid.", __func__), HCCL_E_INTERNAL);
169 :
170 6 : needImport = false;
171 6 : handle = 0;
172 6 : handlePtr = nullptr;
173 6 : tpn = 0;
174 6 : for (const auto& remoteCtx : inner_.remoteJettys) {
175 2 : if (!IsSameRemoteJetty(remoteCtx, remoteQpKey, keySize)) {
176 0 : continue;
177 : }
178 2 : if (remoteCtx.ready) {
179 1 : handle = remoteCtx.handle;
180 1 : handlePtr = remoteCtx.handlePtr;
181 1 : tpn = remoteCtx.tpn;
182 1 : HCCL_INFO(
183 : "[%s] reuse shared remote jetty, handle[%llu], tpn[%u].", __func__,
184 : static_cast<unsigned long long>(handle), tpn);
185 : }
186 2 : return HCCL_SUCCESS;
187 : }
188 :
189 4 : SharedRemoteJettyCtx remoteCtx;
190 4 : remoteCtx.remoteQpKey.assign(remoteQpKey, remoteQpKey + keySize);
191 4 : inner_.remoteJettys.push_back(std::move(remoteCtx));
192 4 : needImport = true;
193 4 : HCCL_INFO("[%s] reserve shared remote jetty import.", __func__);
194 4 : return HCCL_SUCCESS;
195 7 : }
196 :
197 7 : HcclResult JettyContext::PublishSharedRemoteJetty(
198 : const uint8_t* remoteQpKey, uint32_t keySize, uint64_t handle, void* handlePtr, uint32_t tpn)
199 : {
200 7 : CHK_PTR_NULL(remoteQpKey);
201 6 : CHK_PTR_NULL(handlePtr);
202 5 : CHK_PRT_RET(
203 : keySize == 0 || keySize > Hccl::HRT_UB_QP_KEY_MAX_LEN || handle == 0,
204 : HCCL_ERROR(
205 : "[%s] invalid params, keySize[%u], handle[%llu].", __func__, keySize,
206 : static_cast<unsigned long long>(handle)),
207 : HCCL_E_PARA);
208 :
209 3 : std::lock_guard<std::mutex> lk(mtx_);
210 3 : for (auto& remoteCtx : inner_.remoteJettys) {
211 2 : if (!IsSameRemoteJetty(remoteCtx, remoteQpKey, keySize)) {
212 0 : continue;
213 : }
214 2 : remoteCtx.handle = handle;
215 2 : remoteCtx.handlePtr = handlePtr;
216 2 : remoteCtx.tpn = tpn;
217 2 : remoteCtx.ready = true;
218 2 : HCCL_INFO(
219 : "[%s] publish shared remote jetty, handle[%llu], tpn[%u].", __func__,
220 : static_cast<unsigned long long>(handle), tpn);
221 2 : return HCCL_SUCCESS;
222 : }
223 1 : HCCL_ERROR("[%s] shared remote jetty reservation not found.", __func__);
224 1 : return HCCL_E_NOT_FOUND;
225 3 : }
226 :
227 5 : void JettyContext::UnimportSharedRemoteJettys(Inner& inner)
228 : {
229 5 : if (inner.rdmaHandle == nullptr) {
230 4 : return;
231 : }
232 1 : for (auto& remoteCtx : inner.remoteJettys) {
233 0 : if (!remoteCtx.ready || remoteCtx.handle == 0) {
234 0 : continue;
235 : }
236 0 : Hccl::HrtRaUbUnimportJetty(
237 0 : static_cast<Hccl::RdmaHandle>(inner.rdmaHandle), static_cast<Hccl::TargetJettyHandle>(remoteCtx.handle));
238 0 : HCCL_INFO(
239 : "[JettyContext][%s] unimport shared remote jetty, handle[%llu].", __func__,
240 : static_cast<unsigned long long>(remoteCtx.handle));
241 0 : remoteCtx.handle = 0;
242 0 : remoteCtx.handlePtr = nullptr;
243 0 : remoteCtx.ready = false;
244 : }
245 : }
246 :
247 5 : void JettyContext::DestroyJettyResources(Inner& inner)
248 : {
249 5 : if (inner.handle != 0) {
250 1 : Hccl::HrtRaUbDestroyJetty(inner.handle);
251 1 : HCCL_INFO(
252 : "[JettyContext][%s] destroyed shared jetty, handle[%llu].", __func__,
253 : static_cast<unsigned long long>(inner.handle));
254 : }
255 5 : if (inner.jfcHandle != 0 && inner.rdmaHandle != nullptr) {
256 1 : Hccl::HrtRaUbDestroyJfc(static_cast<Hccl::RdmaHandle>(inner.rdmaHandle), inner.jfcHandle);
257 1 : HCCL_INFO(
258 : "[JettyContext][%s] destroyed shared jfc, jfcHandle[%llu].", __func__,
259 : static_cast<unsigned long long>(inner.jfcHandle));
260 : }
261 5 : if (inner.sqPiPtr != nullptr) {
262 5 : (void)hrtFree(inner.sqPiPtr);
263 5 : inner.sqPiPtr = nullptr;
264 : }
265 5 : if (inner.sqCiPtr != nullptr) {
266 5 : (void)hrtFree(inner.sqCiPtr);
267 5 : inner.sqCiPtr = nullptr;
268 : }
269 5 : if (inner.cqPiPtr != nullptr) {
270 5 : (void)hrtFree(inner.cqPiPtr);
271 5 : inner.cqPiPtr = nullptr;
272 : }
273 5 : if (inner.cqCiPtr != nullptr) {
274 5 : (void)hrtFree(inner.cqCiPtr);
275 5 : inner.cqCiPtr = nullptr;
276 : }
277 : // 防御性清空：销毁完成后句柄类字段不再有效，避免后续误用悬挂句柄
278 : // （Release 路径随后会 inner_ = Inner{}，此处清空主要保护 memcpy_s 失败分支等提前销毁场景）
279 5 : inner.handle = 0;
280 5 : inner.jfcHandle = 0;
281 5 : inner.rdmaHandle = nullptr;
282 5 : }
283 :
284 25 : JettyContext::Ctx JettyContext::InnerToCtx(const Inner& inner)
285 : {
286 25 : Ctx ctx{};
287 25 : ctx.handle = inner.handle;
288 25 : ctx.handlePtr = inner.handlePtr;
289 25 : ctx.jettyId = inner.jettyId;
290 25 : ctx.sqBuffVa = inner.sqBuffVa;
291 25 : ctx.dbAddr = inner.dbAddr;
292 25 : ctx.keySize = inner.keySize;
293 25 : ctx.sqDepth = inner.sqDepth;
294 25 : ctx.sqPiPtr = inner.sqPiPtr;
295 25 : ctx.sqCiPtr = inner.sqCiPtr;
296 25 : ctx.cqPiPtr = inner.cqPiPtr;
297 25 : ctx.cqCiPtr = inner.cqCiPtr;
298 25 : ctx.queueIndexMemSize = inner.queueIndexMemSize;
299 25 : ctx.rdmaHandle = inner.rdmaHandle;
300 25 : ctx.jfcHandle = inner.jfcHandle;
301 25 : ctx.cqInfo = inner.cqInfo;
302 25 : ctx.localPsn = inner.localPsn;
303 25 : if (inner.keySize > 0 && inner.keySize <= Hccl::HRT_UB_QP_KEY_MAX_LEN) {
304 19 : (void)memcpy_s(ctx.localQpKey, Hccl::HRT_UB_QP_KEY_MAX_LEN, inner.localQpKey, inner.keySize);
305 : }
306 25 : return ctx;
307 : }
308 :
309 : } // namespace hcomm
|
