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 "dev_ub_connection.h"
12 :
13 : #include <cstdlib>
14 :
15 : #include "hccp_ctx.h"
16 : #include "exception_util.h"
17 : #include "rma_conn_exception.h"
18 : #include "rdma_handle_manager.h"
19 : #include "exchange_ub_conn_dto.h"
20 : #include "env_config/env_config_v2.h"
21 :
22 : namespace Hccl {
23 :
24 : constexpr u32 OPBASED_UB_SQ_DEPTH_MAX = 8192;
25 : constexpr u32 UB_SQ_OFFLOAD_DEPTH = 128;
26 : constexpr u32 UB_SQ_WQEBB_SIZE = 64;
27 : constexpr u32 WQE_NUM_PER_SQE = 4; // URMA约束每个SQE包含4个WQEBB
28 : constexpr u32 UB_MAX_TRANS_SIZE = 256 * 1024 * 1024; // UB单次最大传输量256*1024*1024 Byte
29 : constexpr uint32_t kTpAttrRetryTimesInitBit = 0U;
30 : constexpr uint32_t kTpAttrAtBit = 1U;
31 :
32 110 : DevUbConnection::DevUbConnection(
33 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
34 : const bool devUsed, const HrtUbJfcMode jfcMode, const IpAddress& locIpv4Addr, const IpAddress& rmtIpv4Addr, u8 qos,
35 110 : u8 taTimeOut, CommEngine engine, u32 inSqDepth, JettyMode jettyMode)
36 : : RmaConnection(nullptr, RmaConnType::UB),
37 110 : rdmaHandle(rdmaHandle),
38 110 : locAddr(locAddr),
39 110 : rmtAddr(rmtAddr),
40 110 : opMode(opMode),
41 110 : jfcMode(jfcMode),
42 110 : engine_(engine),
43 110 : locIpv4Addr(locIpv4Addr),
44 110 : rmtIpv4Addr(rmtIpv4Addr),
45 110 : rmtEid(rmtAddr.GetEid()),
46 110 : locEid(locAddr.GetEid()),
47 110 : rmtReverseEid(rmtAddr.GetReverseEid()),
48 110 : qos_(qos),
49 110 : devUsed_(devUsed),
50 110 : taTimeOut_(taTimeOut),
51 110 : sqDepth(inSqDepth),
52 220 : jettyMode_(jettyMode)
53 : {
54 330 : HCCL_INFO(
55 : "[DevUbConnection::DevUbConnection] rmtEid=%s, engine=%d, jettyMode=%d", rmtEid.Describe().c_str(),
56 : static_cast<s32>(engine_), static_cast<s32>(jettyMode_));
57 110 : devLogicId = HrtGetDevice();
58 :
59 110 : auto dieIdAndFuncId = RdmaHandleManager::GetInstance().GetDieAndFuncId(rdmaHandle); // 获取dieId和FuncId
60 110 : dieId = dieIdAndFuncId.first;
61 110 : funcId = dieIdAndFuncId.second;
62 :
63 : // EXTERNAL_INJECT 模式：跳过建 JFC/jetty，等外部调 SetSharedJettyFields 填充
64 110 : if (jettyMode_ == JettyMode::EXTERNAL_INJECT) {
65 0 : sqDepth = OPBASED_UB_SQ_DEPTH_MAX;
66 0 : HCCL_INFO("[DevUbConnection][Constructor] EXTERNAL_INJECT mode, skip JFC/Jetty creation.");
67 0 : return;
68 : }
69 :
70 110 : if (engine_ == COMM_ENGINE_AIV) {
71 1 : CreateAivUrmaJfc();
72 109 : } else if (jfcMode == HrtUbJfcMode::USER_CTL) {
73 0 : jfcHandle = RdmaHandleManager::GetInstance().GetJfcHandleAndCqInfo(rdmaHandle, cqInfo_, jfcMode);
74 : } else {
75 109 : jfcHandle = RdmaHandleManager::GetInstance().GetJfcHandle(rdmaHandle, cqInfo_, jfcMode);
76 : }
77 110 : if (sqDepth == UB_SQ_DEPTH_NOT_SET) {
78 108 : sqDepth = OPBASED_UB_SQ_DEPTH_MAX;
79 108 : if (opMode == OpMode::OFFLOAD && !devUsed) {
80 5 : sqDepth = UB_SQ_OFFLOAD_DEPTH;
81 : }
82 : }
83 330 : HCCL_INFO(
84 : "[DevUbConnection][Constructor] sqDepth[%u], opMode[%d], devUsed[%d]", sqDepth, static_cast<s32>(opMode),
85 : devUsed);
86 :
87 110 : if (sqDepth > (UINT32_MAX / UB_SQ_WQEBB_SIZE / WQE_NUM_PER_SQE)) {
88 0 : THROW<InternalException>("integer overflow occurs");
89 : }
90 :
91 110 : if (!devUsed_) {
92 108 : CreateJetty(devUsed_);
93 : } else {
94 6 : HCCL_INFO("[DevUbConnection][Constructor] devUsed: defer CreateJetty until GetTpInfo maps qos.");
95 : }
96 0 : }
97 :
98 2 : DevUbTpConnection::DevUbTpConnection(
99 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
100 : const bool devUsed, const HrtUbJfcMode jfcMode, const IpAddress& locIpv4Addr, const IpAddress& rmtIpv4Addr, u8 qos,
101 2 : u8 taTimeOut, CommEngine engine, u32 sqDepth, JettyMode jettyMode)
102 : : DevUbConnection(
103 : rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode, locIpv4Addr, rmtIpv4Addr, qos, taTimeOut, engine,
104 2 : sqDepth, jettyMode)
105 : {
106 2 : tpProtocol = TpProtocol::TP;
107 2 : }
108 :
109 4 : DevUbCtpConnection::DevUbCtpConnection(
110 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
111 : const bool devUsed, const HrtUbJfcMode jfcMode, const IpAddress& locIpv4Addr, const IpAddress& rmtIpv4Addr, u8 qos,
112 4 : u8 taTimeOut, CommEngine engine, u32 sqDepth, JettyMode jettyMode)
113 : : DevUbConnection(
114 : rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode, locIpv4Addr, rmtIpv4Addr, qos, taTimeOut, engine,
115 4 : sqDepth, jettyMode)
116 : {
117 4 : tpProtocol = TpProtocol::CTP;
118 4 : }
119 :
120 0 : DevUbUboeConnection::DevUbUboeConnection(
121 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
122 : const bool devUsed, const HrtUbJfcMode jfcMode, const IpAddress& locIpv4Addr, const IpAddress& rmtIpv4Addr, u8 qos,
123 0 : u8 taTimeOut, CommEngine engine, u32 sqDepth, JettyMode jettyMode)
124 : : DevUbConnection(
125 : rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode, locIpv4Addr, rmtIpv4Addr, qos, taTimeOut, engine,
126 0 : sqDepth, jettyMode)
127 : {
128 0 : tpProtocol = TpProtocol::UBOE;
129 0 : jettyTimeOut = 16; // UBOE Jetty异步创建超时 hw_val=16 (对应8s)
130 0 : }
131 :
132 1 : DevUbRtpConnection::DevUbRtpConnection(
133 : const RdmaHandle rdmaHandle, const IpAddress& locAddr, const IpAddress& rmtAddr, const OpMode opMode,
134 : const bool devUsed, const HrtUbJfcMode jfcMode, const IpAddress& locAddrEid, const IpAddress& rmtAddrEid,
135 1 : const u8 qos, u8 taTimeOut, CommEngine engine, u32 sqDepth, JettyMode jettyMode)
136 : : DevUbConnection(
137 : rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode, locAddrEid, rmtAddrEid, qos, taTimeOut, engine,
138 1 : sqDepth, jettyMode)
139 : {
140 1 : tpProtocol = TpProtocol::UB_RTP;
141 : // UB_RTP与UBOE同属UB传输，Jetty异步创建超时一致，均为16秒
142 1 : jettyTimeOut = 16;
143 1 : }
144 :
145 47 : std::vector<char> DevUbConnection::GetUniqueId() const
146 : {
147 47 : BinaryStream binaryStream;
148 47 : binaryStream << dieId;
149 47 : binaryStream << funcId;
150 47 : binaryStream << jettyId;
151 :
152 47 : u32 jfcPollMode = 0; // 待修改，0代表STARS POLL，1代表software Poll
153 47 : bool dwqeCacheLocked = false; // 待修改，该jetty是否支持dwqeCachedLocked，默认不支持
154 47 : u64 sqCiAddr = 0; // 待修改，软件poll CQ情况下，需要AICPU从该地址中读取CI,依赖UB驱动支持
155 47 : binaryStream << jfcPollMode;
156 47 : binaryStream << dwqeCacheLocked;
157 47 : binaryStream << dbAddr;
158 47 : binaryStream << sqCiAddr;
159 47 : binaryStream << sqBuffVa;
160 47 : binaryStream << sqDepth;
161 47 : binaryStream << tpn;
162 47 : binaryStream << rmtEid.raw;
163 47 : binaryStream << locEid.raw;
164 47 : binaryStream << maxReadSize;
165 47 : binaryStream << maxWriteSize;
166 47 : binaryStream << static_cast<uint64_t>(jettyHandle);
167 :
168 47 : std::vector<char> result;
169 47 : binaryStream.Dump(result);
170 141 : HCCL_INFO("DevUbConnection::GetUniqueId:%s", Describe().c_str());
171 141 : HCCL_INFO(
172 : "type=%s, jfcPollMode=%u, dwqeCacheLocked=%d, sqCiAddr=0x%llx", rmaConnType.Describe().c_str(), jfcPollMode,
173 : dwqeCacheLocked, sqCiAddr);
174 47 : return result;
175 47 : }
176 :
177 0 : void DevUbConnection::SetCqInfo(HcclAiRMACQ& cq) const
178 : {
179 0 : cq.jfcId = cqInfo_.id;
180 0 : cq.cqVA = cqInfo_.va;
181 0 : cq.cqeSize = cqInfo_.cqeSize;
182 0 : cq.cqDepth = cqInfo_.cqDepth;
183 0 : cq.dbAddr = cqInfo_.swdbAddr;
184 0 : }
185 :
186 2 : void DevUbConnection::SetWqInfo(HcclAiRMAWQ& wq)
187 : {
188 2 : wq.jettyId = jettyId;
189 2 : wq.dbAddr = dbAddr;
190 2 : wq.sqVA = sqBuffVa;
191 2 : wq.sqDepth = sqDepth * WQE_NUM_PER_SQE;
192 2 : wq.tp_id = tpn;
193 2 : errno_t ret = memcpy_s(wq.rmtEid, sizeof(wq.rmtEid), rmtReverseEid.raw, sizeof(wq.rmtEid));
194 2 : if (ret != EOK) {
195 3 : HCCL_ERROR("[DevUbConnection][%s] memcpy_s failed, ret=%d", __func__, ret);
196 2 : ThrowAbnormalStatus(std::string(__func__));
197 : }
198 1 : }
199 :
200 2 : void DevUbConnection::SetSqContextInfo(SqContext& sq)
201 : {
202 2 : sq.contextInfo.ubJfs.jfsID = jettyId;
203 2 : sq.contextInfo.ubJfs.dbVa = dbAddr;
204 2 : sq.contextInfo.ubJfs.sqVa = sqBuffVa;
205 2 : sq.contextInfo.ubJfs.sqDepth = sqDepth * WQE_NUM_PER_SQE;
206 2 : sq.contextInfo.ubJfs.tpID = tpn;
207 4 : errno_t ret = memcpy_s(
208 2 : sq.contextInfo.ubJfs.remoteEID, sizeof(sq.contextInfo.ubJfs.remoteEID), rmtReverseEid.raw,
209 : sizeof(sq.contextInfo.ubJfs.remoteEID));
210 2 : if (ret != EOK) {
211 3 : HCCL_ERROR("[DevUbConnection][%s] memcpy_s failed, ret=%d", __func__, ret);
212 2 : ThrowAbnormalStatus(std::string(__func__));
213 : }
214 1 : }
215 :
216 0 : void DevUbConnection::SetCqContextInfo(CqContext& cq) const
217 : {
218 0 : cq.contextInfo.ubJfc.jfcID = cqInfo_.id;
219 0 : cq.contextInfo.ubJfc.scqVa = cqInfo_.va;
220 0 : cq.contextInfo.ubJfc.cqeSize = cqInfo_.cqeSize;
221 0 : cq.contextInfo.ubJfc.cqDepth = cqInfo_.cqDepth;
222 0 : cq.contextInfo.ubJfc.dbVa = cqInfo_.swdbAddr;
223 0 : }
224 :
225 0 : void DevUbConnection::Connect() { GetStatus(); }
226 :
227 0 : inline uint32_t GetRandomNum()
228 : {
229 0 : uint32_t randNum = std::rand();
230 0 : return randNum;
231 : }
232 :
233 1 : HcclResult DevUbConnection::CalcTotalTimeout(uint32_t& outTotalTimeoutMs)
234 : {
235 1 : TpHandle tpHandle = tpInfo.tpHandle;
236 1 : uint32_t attrBitmap = (1U << kTpAttrRetryTimesInitBit) | (1U << kTpAttrAtBit);
237 1 : struct TpAttr tpAttr = {};
238 1 : u32 devicePhyId = HrtGetDevicePhyIdByIndex(devLogicId);
239 4 : CHK_RET(HrtRaGetTpAttrAsync(devicePhyId, rdmaHandle, tpHandle, attrBitmap, tpAttr, reqHandle));
240 0 : TpAttrInfo tpAttrInfo = TpAttrInfo(tpAttr);
241 0 : CHK_RET(TpManager::GetTpTotalTimeout(tpAttrInfo, outTotalTimeoutMs));
242 0 : return HCCL_SUCCESS;
243 : }
244 :
245 6 : void DevUbConnection::GetTimeOut() // 基于调用方按协议从环境变量获取并传入的超时值控制
246 : {
247 6 : if (tpProtocol == TpProtocol::INVALID) { // 不感知tp建链，当前默认不支持
248 0 : HCCL_ERROR(
249 : "[DevUbConnection][%s] failed, tpProtocol[%s] is not expected.", __func__, tpProtocol.Describe().c_str());
250 0 : ThrowAbnormalStatus(std::string(__func__));
251 0 : return;
252 : }
253 :
254 : // CTP 协议不感知 TP 建链，跳过 CalcTotalTimeout，
255 : // tpTimeOutMs 保持 0，由 TpManager::CalcTaTimeout 内部按 CTP 规则直接使用 taTimeOut_
256 6 : uint32_t tpTimeOutMs = 0;
257 6 : if (tpProtocol != TpProtocol::CTP) {
258 1 : HcclResult ret = CalcTotalTimeout(tpTimeOutMs);
259 1 : if (ret != HCCL_SUCCESS) {
260 3 : HCCL_RUN_WARNING(
261 : "[DevUbConnection][%s] CalcTotalTimeout failed[%d], tpTimeOutMs remains 0.", __func__, ret);
262 : }
263 : }
264 6 : jettyTimeOut = TpManager::CalcTaTimeout(tpProtocol, taTimeOut_, tpTimeOutMs);
265 18 : HCCL_INFO(
266 : "[DevUbConnection][%s] final TA Timeout [%u] (%ums).", __func__, jettyTimeOut,
267 : TpManager::TaHwValueToMs(jettyTimeOut));
268 : }
269 :
270 : /*
271 : * UB 建链状态机（GetTpInfo/CreateJetty 异步未完成则停；同步成功时可同次推进）:
272 : * INIT --GetTpInfo fail--> TP_INFO_GETTING
273 : * INIT / TP_INFO_GETTING --GetTpInfo ok--> CreateJetty --> JETTY_CREATING | JETTY_CREATED
274 : * （EXTERNAL_INJECT 模式时跳过 CreateJetty，直接 JETTY_CREATED）
275 : * JETTY_CREATING --create done--> JETTY_CREATED (EXCHANGEABLE)
276 : * JETTY_CREATED --ImportRmtDto--> JETTY_IMPORTING（此处不推进）
277 : * JETTY_IMPORTING --import done--> READY
278 : */
279 4 : void DevUbConnection::AdvanceUbConnFromJettyImporting()
280 : {
281 4 : SetImportInfo();
282 :
283 4 : if (jettyMode_ == JettyMode::EXTERNAL_INJECT) {
284 0 : if (publishRemoteCb_ == nullptr) {
285 0 : THROW<InternalException>("[DevUbConnection][%s] publish callback is null.", __func__);
286 : }
287 0 : HcclResult ret = publishRemoteCb_(remoteQpKey, keySize, remoteJettyHandle, remoteJettyHandlePtr, tpn);
288 0 : if (ret != HCCL_SUCCESS) {
289 0 : THROW<InternalException>(
290 : "[DevUbConnection][%s] publish shared remote jetty failed, ret[%d].", __func__, ret);
291 : }
292 : }
293 :
294 4 : status = RmaConnStatus::READY;
295 4 : ubConnStatus = UbConnStatus::READY;
296 4 : }
297 :
298 0 : void DevUbConnection::AdvanceUbConnFromJettyImportWaiting()
299 : {
300 : // 超时检查：超过 jettyTimeOut 秒仍未 publish，则对端异常，抛异常避免永久阻塞。
301 : // 每次轮询都检查，保证超时判定不被退避延迟。
302 : auto elapsedSec
303 0 : = std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - importWaitingStart_);
304 0 : if (elapsedSec.count() >= static_cast<int64_t>(jettyTimeOut)) {
305 0 : HCCL_ERROR(
306 : "[DevUbConnection][%s] JETTY_IMPORT_WAITING timeout[%us], remote jetty not published.", __func__,
307 : static_cast<uint32_t>(elapsedSec.count()));
308 0 : THROW<InternalException>("[DevUbConnection][%s] wait shared remote jetty publish timeout.");
309 : }
310 :
311 : // 退避：每 8 次 GetStatus 轮询才真正调 acquireRemoteCb_ 一次，减少 JettyContext 锁竞争。
312 : // 首次进入立即调用（pollCount==0），后续每 8 次轮询触发一次实际查询。
313 0 : if (importWaitingPollCount_ > 0 && (importWaitingPollCount_ % 8U) != 0U) {
314 0 : importWaitingPollCount_++;
315 0 : return;
316 : }
317 0 : importWaitingPollCount_++;
318 :
319 0 : AcquireOrWaitSharedRemoteJetty();
320 : }
321 :
322 20 : RmaConnStatus DevUbConnection::GetStatus()
323 : {
324 : // 稳定态 / 等待外部 ImportRmtDto：无需推进
325 20 : if (ubConnStatus == UbConnStatus::READY || ubConnStatus == UbConnStatus::JETTY_CREATED) {
326 2 : return status;
327 : }
328 :
329 18 : if (!CheckRequestResult()) {
330 0 : return status;
331 : }
332 :
333 18 : switch (ubConnStatus) {
334 6 : case UbConnStatus::INIT:
335 6 : ProcessInit();
336 6 : break;
337 6 : case UbConnStatus::TP_INFO_GETTING:
338 6 : if (!GetTpInfo()) {
339 0 : break;
340 : }
341 6 : ProcessCreateJetty();
342 6 : break;
343 1 : case UbConnStatus::JETTY_CREATING:
344 1 : SetJettyInfo();
345 1 : status = RmaConnStatus::EXCHANGEABLE;
346 1 : ubConnStatus = UbConnStatus::JETTY_CREATED;
347 1 : break;
348 4 : case UbConnStatus::JETTY_IMPORTING:
349 4 : AdvanceUbConnFromJettyImporting();
350 4 : break;
351 0 : case UbConnStatus::JETTY_IMPORT_WAITING:
352 0 : AdvanceUbConnFromJettyImportWaiting();
353 0 : break;
354 0 : case UbConnStatus::READY:
355 0 : break;
356 1 : default:
357 2 : ThrowAbnormalStatus(std::string(__func__));
358 0 : break;
359 : }
360 :
361 17 : return status;
362 : }
363 :
364 6 : void DevUbConnection::ProcessInit()
365 : {
366 18 : HCCL_INFO(
367 : "[DevUbConnection][%s] start, status[%s], ubConnStatus[%s].", __func__, status.Describe().c_str(),
368 : ubConnStatus.Describe().c_str());
369 6 : if (!GetTpInfo()) {
370 6 : ubConnStatus = UbConnStatus::TP_INFO_GETTING;
371 6 : return;
372 : }
373 0 : ProcessCreateJetty();
374 : }
375 :
376 6 : void DevUbConnection::ProcessCreateJetty()
377 : {
378 6 : GetTimeOut();
379 6 : if (jettyMode_ == JettyMode::EXTERNAL_INJECT) {
380 : // 共享 jetty：句柄已由 SetSharedJettyFields 注入，跳过 CreateJetty
381 0 : status = RmaConnStatus::EXCHANGEABLE;
382 0 : ubConnStatus = UbConnStatus::JETTY_CREATED;
383 0 : HCCL_INFO("[DevUbConnection][%s] shared jetty mode, skip CreateJetty, direct to JETTY_CREATED.", __func__);
384 0 : return;
385 : }
386 6 : CreateJetty(devUsed_);
387 6 : if (devUsed_ || !CheckRequestResult()) {
388 1 : ubConnStatus = UbConnStatus::JETTY_CREATING;
389 1 : return;
390 : }
391 5 : SetJettyInfo();
392 5 : status = RmaConnStatus::EXCHANGEABLE;
393 5 : ubConnStatus = UbConnStatus::JETTY_CREATED;
394 : }
395 :
396 6 : std::unique_ptr<Serializable> DevUbConnection::GetExchangeDto()
397 : {
398 6 : if (status != RmaConnStatus::READY && status != RmaConnStatus::EXCHANGEABLE) {
399 0 : HCCL_ERROR("[DevUbConnection][%s] status[%s] is not expected.", __func__, status.Describe().c_str());
400 0 : ThrowAbnormalStatus(std::string(__func__));
401 : }
402 :
403 6 : if (tpProtocol != TpProtocol::INVALID) {
404 4 : jettyImportCfg.localTpHandle = tpInfo.tpHandle;
405 :
406 12 : HCCL_INFO(
407 : "[DevUbConnection][%s] tpEnable, localTpHandle[0x%llx] localPsn[%u].", __func__,
408 : jettyImportCfg.localTpHandle, jettyImportCfg.localPsn);
409 : }
410 :
411 : std::unique_ptr<ExchangeUbConnDto> dto
412 6 : = make_unique<ExchangeUbConnDto>(tokenValue, keySize, jettyImportCfg.localTpHandle, jettyImportCfg.localPsn);
413 6 : (void)memcpy_s(dto->qpKey, HRT_UB_QP_KEY_MAX_LEN, localQpKey, HRT_UB_QP_KEY_MAX_LEN);
414 12 : return std::unique_ptr<Serializable>(dto.release());
415 6 : }
416 :
417 4 : void DevUbConnection::ParseRmtExchangeDto(const Serializable& rmtDto)
418 : {
419 4 : auto dto = dynamic_cast<const ExchangeUbConnDto&>(rmtDto);
420 12 : HCCL_INFO("[DevUbConnection][%s] remoteConnDto[%s]", __func__, dto.Describe().c_str());
421 4 : remoteTokenValue = dto.tokenValue;
422 4 : (void)memcpy_s(remoteQpKey, HRT_UB_QP_KEY_MAX_LEN, dto.qpKey, HRT_UB_QP_KEY_MAX_LEN);
423 :
424 4 : if (tpProtocol != TpProtocol::INVALID) {
425 4 : jettyImportCfg.remoteTpHandle = dto.tpHandle;
426 4 : jettyImportCfg.remotePsn = dto.psn;
427 12 : HCCL_INFO(
428 : "[DevUbConnection][%s] tpEnable, remoteTpHandle[0x%llx], remotePsn[%u].", __func__,
429 : jettyImportCfg.remoteTpHandle, jettyImportCfg.remotePsn);
430 : }
431 4 : }
432 :
433 6 : void DevUbConnection::ImportRmtDto()
434 : {
435 6 : if (ubConnStatus == UbConnStatus::READY) {
436 3 : HCCL_WARNING("[DevUbConnection][%s] import jetty already, %s.", __func__, Describe().c_str());
437 1 : return;
438 : }
439 :
440 5 : if (ubConnStatus != UbConnStatus::JETTY_CREATED) {
441 0 : HCCL_ERROR(
442 : "[DevUbConnection][%s] failed, ubConnStatus[%s] is not expected.", __func__,
443 : ubConnStatus.Describe().c_str());
444 0 : ThrowAbnormalStatus(std::string(__func__));
445 : }
446 :
447 5 : if (jettyMode_ == JettyMode::EXTERNAL_INJECT) {
448 0 : AcquireOrWaitSharedRemoteJetty();
449 0 : return;
450 : }
451 :
452 5 : ImportJetty();
453 5 : ubConnStatus = UbConnStatus::JETTY_IMPORTING;
454 : }
455 :
456 4 : void DevUbConnection::ThrowAbnormalStatus(std::string funcName)
457 : {
458 4 : auto errMsg = StringFormat("[DevUbConnection][%s] failed, [%s].", funcName.c_str(), Describe().c_str());
459 4 : status = RmaConnStatus::CONN_INVALID;
460 4 : ubConnStatus = UbConnStatus::CONN_INVALID;
461 4 : THROW<RmaConnException>(errMsg);
462 4 : }
463 :
464 23 : bool DevUbConnection::CheckRequestResult()
465 : {
466 23 : if (reqHandle == 0) {
467 7 : return true;
468 : }
469 :
470 16 : ReqHandleResult result = HrtRaGetAsyncReqResult(reqHandle);
471 16 : if (result == ReqHandleResult::NOT_COMPLETED) {
472 0 : return false;
473 : }
474 :
475 16 : if (result != ReqHandleResult::COMPLETED) {
476 0 : THROW<InternalException>(
477 0 : "[DevUbConnection][%s] failed, result[%s] is unexpected.", __func__, result.Describe().c_str());
478 : }
479 :
480 16 : return true;
481 : }
482 :
483 116 : void DevUbConnection::CreateJetty(const bool devUsed)
484 : {
485 116 : if (sqDepth > UINT32_MAX / UB_SQ_WQEBB_SIZE / WQE_NUM_PER_SQE) {
486 0 : THROW<InternalException>(
487 : "[DevUbConnection][%s] failed, sqDepth[%u] times "
488 : "UB_SQ_WQEBB_SIZE[%u] overflow uint32 max.",
489 : __func__, sqDepth, UB_SQ_WQEBB_SIZE);
490 : }
491 116 : u32 size = static_cast<u32>(sqDepth) * static_cast<u32>(UB_SQ_WQEBB_SIZE) * static_cast<u32>(WQE_NUM_PER_SQE);
492 : HrtRaUbCreateJettyParam req{
493 : jfcHandle,
494 : jfcHandle,
495 : GetUbToken(),
496 : 0,
497 : HrtJettyMode::HOST_OPBASE, // 默认HOST单算子模式
498 : 0, // HOST展开与AICPU展开传入jetty id为0，申请一个新的jetty
499 : 0, // va由底层分配，此处填0即可。
500 : size,
501 : 0,
502 : sqDepth,
503 116 : jettyTimeOut}; // 非CCUv2不需要填写sqeBufIndex
504 :
505 116 : if (opMode == OpMode::OFFLOAD) { // HOST展开图模式切换模式
506 5 : req.jettyMode = HrtJettyMode::HOST_OFFLOAD;
507 : }
508 :
509 116 : if (devUsed) { // AICPU场景切换模式
510 1 : req.jettyMode = HrtJettyMode::DEV_USED;
511 3 : HCCL_INFO("[DevUbConnection][%s] HrtJettyMode is DEV_USED.", __func__);
512 : }
513 :
514 116 : if (tpInfo.hasMappedJettyPriority) {
515 2 : req.qos = static_cast<u8>(tpInfo.mappedJettyPriority & 0xFU);
516 : }
517 348 : HCCL_INFO(
518 : "[DevUbConnection][%s] jetty create qos[%u] (maps to attr.ub.priority lower 4 bits).", __func__,
519 : static_cast<unsigned int>(req.qos));
520 :
521 116 : reqHandle = RaUbCreateJettyAsync(rdmaHandle, req, reqDataBuffer, jettyHandlePtr);
522 116 : }
523 :
524 0 : HcclResult DevUbConnection::SetSharedJettyFields(
525 : JettyHandle jettyHdl, void* jettyHdlPtr, uint32_t jId, uint64_t sqVa, uint64_t db, const uint8_t* qpKey,
526 : uint32_t kSize, uint32_t sDepth, JfcHandle sharedJfc, CqCreateInfo sharedCqInfo, uint32_t sharedLocalPsn,
527 : void* epTag, std::function<void(void*)> releaseCb, AcquireSharedRemoteJettyCallback acquireRemoteCb,
528 : PublishSharedRemoteJettyCallback publishRemoteCb)
529 : {
530 0 : if (jettyHdl == 0 || jettyHdlPtr == nullptr || sDepth == 0 || acquireRemoteCb == nullptr
531 0 : || publishRemoteCb == nullptr) {
532 0 : HCCL_ERROR(
533 : "[DevUbConnection][%s] invalid params, jettyHdl[0x%llx], jettyHdlPtr[%p], sDepth[%u], "
534 : "acquireRemoteCb[%d], publishRemoteCb[%d].",
535 : __func__, static_cast<unsigned long long>(jettyHdl), jettyHdlPtr, sDepth,
536 : acquireRemoteCb == nullptr ? 0 : 1, publishRemoteCb == nullptr ? 0 : 1);
537 0 : return HCCL_E_PARA;
538 : }
539 0 : if (jettyMode_ != JettyMode::EXTERNAL_INJECT) {
540 0 : HCCL_ERROR(
541 : "[DevUbConnection][%s] not EXTERNAL_INJECT mode, jettyMode[%d], reject SetSharedJettyFields.", __func__,
542 : static_cast<s32>(jettyMode_));
543 0 : return HCCL_E_INTERNAL;
544 : }
545 : // EXTERNAL_INJECT 模式构造时跳过了建 JFC/jetty，此处填充共享字段，无需销毁私有 JFC
546 0 : if (qpKey != nullptr && kSize > 0 && kSize <= HRT_UB_QP_KEY_MAX_LEN) {
547 0 : s32 ret = memcpy_s(&localQpKey[0], HRT_UB_QP_KEY_MAX_LEN, qpKey, kSize);
548 0 : if (ret != EOK) {
549 0 : HCCL_ERROR("[DevUbConnection][%s] memcpy_s localQpKey failed, ret[%d].", __func__, ret);
550 0 : return HCCL_E_INTERNAL;
551 : }
552 : }
553 0 : endpointTag_ = epTag;
554 0 : releaseCb_ = std::move(releaseCb);
555 0 : acquireRemoteCb_ = std::move(acquireRemoteCb);
556 0 : publishRemoteCb_ = std::move(publishRemoteCb);
557 : // 共享 jetty 模式下主 connection 仍各自调 GetTpInfo 申请本 pair 的 tpHandle（TpManager useCnt++），
558 : // 析构时必须 ReleaseTp 释放引用，否则 useCnt 泄漏导致 tpHandle 永不回收。
559 0 : releaseTpOnDestroy_ = true;
560 0 : jettyHandle = jettyHdl;
561 0 : jettyHandlePtr = jettyHdlPtr;
562 0 : jettyId = jId;
563 0 : sqBuffVa = sqVa;
564 0 : dbAddr = db;
565 0 : keySize = kSize;
566 0 : sqDepth = sDepth;
567 0 : jfcHandle = sharedJfc;
568 0 : cqInfo_ = sharedCqInfo;
569 : // 注入临时 connection 生成的 localPsn，使主 connection 的 GetExchangeDto 发送与共享 jetty
570 : // 一致的 psn。多个主 connection 共享同一本地 jetty/SQ，必须用同一 localPsn，避免各自
571 : // GenerateLocalPsn 生成不同 psn 后 import 同一 TP 对时 psn 互相覆盖导致硬件传输错乱。
572 : // 注意：tpHandle 不在此注入——一对多场景下各主 connection 到不同对端需各自向管控面申请
573 : // 自己的 tpHandle，否则对端 import 时 peerTpHandle 路由不匹配。
574 0 : jettyImportCfg.localPsn = sharedLocalPsn;
575 : // 注入后不直接跳状态机：仍需走 GetTpInfo 申请本 pair 的 TP，由 ProcessCreateJetty
576 : // 中 EXTERNAL_INJECT 分支跳过 CreateJetty 直接进入 JETTY_CREATED
577 0 : HCCL_INFO(
578 : "[DevUbConnection][%s] shared jetty fields set, handle[0x%llx], jettyId[%u], sqDepth[%u], "
579 : "jfcHandle[%llu].",
580 : __func__, static_cast<unsigned long long>(jettyHandle), jettyId, sqDepth,
581 : static_cast<unsigned long long>(jfcHandle));
582 0 : return HCCL_SUCCESS;
583 : }
584 :
585 0 : void DevUbConnection::DetachJetty()
586 : {
587 : // SELF_CREATE 模式建好 jetty 后调用：分离 jetty 所有权，析构不销毁 jetty，交由 JettyContext 管理。
588 : // 注意：不置 releaseTpOnDestroy_=false——临时 connection 自己申请的 TP 引用仍需在析构时
589 : // ReleaseTp 释放（TpManager 引用计数 -1），否则 useCnt 泄漏。
590 : // 共享 jetty 模式下主 connection 不复用临时 connection 的 tpHandle：一对多场景各主 connection
591 : // 到不同对端需各自向 TpManager 申请本 pair 的 tpHandle，否则对端 import 时 peerTpHandle 路由不匹配。
592 : // 临时 connection 的 tpHandle 引用随析构释放，不影响主 connection 各自申请的 tpHandle。
593 0 : jettyDetached_ = true;
594 0 : HCCL_INFO(
595 : "[DevUbConnection][%s] jetty ownership detached, handle[0x%llx].", __func__,
596 : static_cast<unsigned long long>(jettyHandle));
597 0 : }
598 :
599 0 : HcclResult DevUbConnection::GetJettyInfo(JettyInfo& info) const
600 : {
601 0 : info.handle = jettyHandle;
602 0 : info.handlePtr = jettyHandlePtr;
603 0 : info.jettyId = jettyId;
604 0 : info.sqBuffVa = sqBuffVa;
605 0 : info.dbAddr = dbAddr;
606 0 : info.keySize = keySize;
607 0 : info.sqDepth = sqDepth;
608 0 : info.rdmaHandle = rdmaHandle;
609 0 : info.jfcHandle = jfcHandle;
610 0 : info.cqInfo = cqInfo_;
611 0 : info.localPsn = jettyImportCfg.localPsn;
612 0 : auto sRet = memcpy_s(&info.localQpKey[0], HRT_UB_QP_KEY_MAX_LEN, localQpKey, HRT_UB_QP_KEY_MAX_LEN);
613 0 : if (sRet != EOK) {
614 0 : HCCL_ERROR("[DevUbConnection][%s] memcpy_s failed, ret[%d].", __func__, sRet);
615 0 : return HCCL_E_INTERNAL;
616 : }
617 0 : return HCCL_SUCCESS;
618 : }
619 :
620 7 : void DevUbConnection::SetJettyInfo()
621 : {
622 7 : struct QpCreateInfo* info = reinterpret_cast<QpCreateInfo*>(reqDataBuffer.data());
623 7 : jettyId = info->ub.id;
624 7 : jettyHandle = reinterpret_cast<JettyHandle>(jettyHandlePtr);
625 7 : keySize = info->key.size;
626 7 : sqBuffVa = info->ub.sqBuffVa; // hccp提供
627 21 : HCCL_RUN_INFO(
628 : "[DevUbConnection][%s] Get sqBuffVa is %llx. jettyId[%u], jettyHandle[%llx], dieId[%u], funcId[%u]", __func__,
629 : sqBuffVa, jettyId, jettyHandle, dieId, funcId);
630 :
631 7 : s32 ret = memcpy_s(&localQpKey[0], HRT_UB_QP_KEY_MAX_LEN, info->key.value, info->key.size);
632 7 : if (ret != 0) {
633 0 : THROW<InternalException>(StringFormat("[DevUbConnection][%s] memcpy_s failed, ret=%d", __func__, ret));
634 : }
635 :
636 7 : dbAddr = info->ub.dbAddr;
637 7 : }
638 :
639 12 : bool DevUbConnection::GetTpInfo()
640 : {
641 12 : if (tpProtocol == TpProtocol::INVALID) { // 不感知tp建链，当前默认不支持
642 0 : HCCL_ERROR(
643 : "[DevUbConnection][%s] failed, tpProtocol[%s] is not expected.", __func__, tpProtocol.Describe().c_str());
644 0 : ThrowAbnormalStatus(std::string(__func__));
645 : }
646 :
647 : // 共享 jetty 模式：tpHandle 不复用临时 connection 的值——一对多场景下各主 connection 到不同对端
648 : // 必须各自向管控面申请本 pair 的 tpHandle，否则对端 import 时 peerTpHandle 路由不匹配。
649 : // 但 localPsn 已由 SetSharedJettyFields 注入（来自临时 connection），不在此 GenerateLocalPsn，
650 : // 避免多主 connection 共享同一本地 jetty/SQ 各自生成不同 psn 后 import 同一 TP 对时互相覆盖。
651 12 : const bool isSharedJettyMode = (jettyMode_ == JettyMode::EXTERNAL_INJECT);
652 :
653 12 : RaUbGetTpInfoParam p{};
654 12 : p.locAddr = locAddr;
655 12 : p.rmtAddr = rmtAddr;
656 12 : p.tpProtocol = tpProtocol;
657 12 : p.qos = static_cast<uint32_t>(qos_);
658 12 : p.slLevelCount = 0;
659 12 : p.loopFirstTpLowestSl = false;
660 12 : p.locIpv4Addr = locIpv4Addr;
661 12 : p.rmtIpv4Addr = rmtIpv4Addr;
662 :
663 12 : auto ret = TpManager::GetInstance(devLogicId).GetTpInfo(p, tpInfo);
664 :
665 12 : switch (ret) {
666 6 : case HcclResult::HCCL_SUCCESS:
667 6 : if (isSharedJettyMode) {
668 0 : HCCL_INFO(
669 : "[DevUbConnection][%s] shared jetty mode, apply own tpHandle[0x%llx] for rmtAddr[%s], "
670 : "reuse injected localPsn[%u].",
671 : __func__, static_cast<unsigned long long>(tpInfo.tpHandle), rmtAddr.Describe().c_str(),
672 : jettyImportCfg.localPsn);
673 : } else {
674 6 : GenerateLocalPsn();
675 : }
676 6 : return true;
677 6 : case HcclResult::HCCL_E_AGAIN:
678 6 : return false;
679 0 : case HcclResult::HCCL_E_NOT_FOUND:
680 : default:
681 0 : HCCL_ERROR("[DevUbConnection][%s] failed, hccl result[%d]", __func__, ret);
682 0 : ThrowAbnormalStatus(std::string(__func__));
683 0 : break;
684 : }
685 0 : return true;
686 : }
687 :
688 6 : void DevUbConnection::GenerateLocalPsn() { jettyImportCfg.localPsn = GetRandomNum(); }
689 :
690 5 : void DevUbConnection::ImportJetty()
691 : {
692 5 : HrtRaUbJettyImportedInParam in{};
693 5 : in.key = remoteQpKey;
694 5 : in.keyLen = keySize;
695 5 : in.tokenValue = remoteTokenValue;
696 5 : in.jettyImportCfg = jettyImportCfg;
697 5 : in.jettyImportCfg.protocol = tpProtocol;
698 :
699 7 : if (tpProtocol != TpProtocol::CTP && tpProtocol != TpProtocol::TP && tpProtocol != TpProtocol::UBOE
700 7 : && tpProtocol != TpProtocol::UB_RTP) {
701 0 : HCCL_ERROR(
702 : "[DevUbConnection][%s] failed, tp protocol[%s] is not expected, %s.", __func__,
703 : tpProtocol.Describe().c_str(), Describe().c_str());
704 0 : ThrowAbnormalStatus(std::string(__func__));
705 : }
706 :
707 5 : reqHandle = RaUbTpImportJettyAsync(rdmaHandle, in, reqDataBuffer, remoteJettyHandlePtr);
708 5 : }
709 :
710 4 : void DevUbConnection::SetImportInfo()
711 : {
712 4 : struct QpImportInfoT* info = reinterpret_cast<QpImportInfoT*>(reqDataBuffer.data());
713 4 : remoteJettyHandle = reinterpret_cast<TargetJettyHandle>(remoteJettyHandlePtr);
714 4 : tpn = info->out.ub.tpn;
715 4 : }
716 :
717 0 : void DevUbConnection::SetSharedRemoteJettyInfo(TargetJettyHandle handle, void* handlePtr, uint32_t remoteTpn)
718 : {
719 0 : remoteJettyHandle = handle;
720 0 : remoteJettyHandlePtr = handlePtr;
721 0 : tpn = remoteTpn;
722 0 : status = RmaConnStatus::READY;
723 0 : ubConnStatus = UbConnStatus::READY;
724 0 : HCCL_INFO(
725 : "[DevUbConnection][%s] reuse shared remote jetty, handle[0x%llx], tpn[%u].", __func__,
726 : static_cast<unsigned long long>(handle), remoteTpn);
727 0 : }
728 :
729 0 : void DevUbConnection::AcquireOrWaitSharedRemoteJetty()
730 : {
731 0 : if (acquireRemoteCb_ == nullptr) {
732 0 : THROW<InternalException>("[DevUbConnection][%s] acquire callback is null.", __func__);
733 : }
734 :
735 0 : bool needImport = false;
736 0 : TargetJettyHandle cachedHandle = 0;
737 0 : void* cachedHandlePtr = nullptr;
738 0 : uint32_t cachedTpn = 0;
739 0 : HcclResult ret = acquireRemoteCb_(remoteQpKey, keySize, needImport, cachedHandle, cachedHandlePtr, cachedTpn);
740 0 : if (ret != HCCL_SUCCESS) {
741 0 : THROW<InternalException>("[DevUbConnection][%s] acquire shared remote jetty failed, ret[%d].", __func__, ret);
742 : }
743 :
744 0 : if (cachedHandle != 0) {
745 0 : SetSharedRemoteJettyInfo(cachedHandle, cachedHandlePtr, cachedTpn);
746 0 : return;
747 : }
748 0 : if (needImport) {
749 0 : ImportJetty();
750 0 : ubConnStatus = UbConnStatus::JETTY_IMPORTING;
751 0 : HCCL_INFO("[DevUbConnection][%s] start shared remote jetty import.", __func__);
752 0 : return;
753 : }
754 : // 进入 WAITING：记录起始时刻，供 AdvanceUbConnFromJettyImportWaiting 做超时判断
755 0 : if (ubConnStatus != UbConnStatus::JETTY_IMPORT_WAITING) {
756 0 : importWaitingStart_ = std::chrono::steady_clock::now();
757 0 : importWaitingPollCount_ = 0;
758 : }
759 0 : ubConnStatus = UbConnStatus::JETTY_IMPORT_WAITING;
760 : }
761 :
762 114 : void DevUbConnection::ReleaseTp()
763 : {
764 114 : ReleaseUbConnectionTp(devLogicId, locAddr, rmtAddr, tpProtocol, tpInfo, static_cast<uint32_t>(qos_));
765 114 : }
766 :
767 112 : void DevUbConnection::ReleaseRemoteJettyIfImported(bool ctxValid)
768 : {
769 : // EXTERNAL_INJECT 模式（主 connection）：远端 jetty 由 JettyContext 统一 unimport，不在此处理
770 112 : if (jettyMode_ == JettyMode::EXTERNAL_INJECT) {
771 0 : return;
772 : }
773 112 : if (remoteJettyHandle == 0) {
774 112 : return;
775 : }
776 0 : if (!ctxValid) {
777 0 : HCCL_WARNING(
778 : "[DevUbConnection][%s] skip HrtRaUbUnimportJetty, "
779 : "rdmaHandle=%p invalid (DeInit/DestroyAll done), remoteJettyHandle=0x%llx",
780 : __func__, rdmaHandle, static_cast<unsigned long long>(remoteJettyHandle));
781 : } else {
782 0 : HrtRaUbUnimportJetty(rdmaHandle, remoteJettyHandle);
783 : }
784 0 : remoteJettyHandle = 0;
785 : }
786 :
787 0 : void DevUbConnection::ReleaseSharedJettyModeResources(bool ctxValid)
788 : {
789 : // EXTERNAL_INJECT 模式（主 connection）：构造时跳过 JFC/jetty 创建，jetty + JFC + CQ 全部由
790 : // JettyContext 统一管理，connection 不销毁。通过 releaseCb_ 通知 Endpoint 减引用计数
791 : // （refCount 归 0 时 JettyContext 销毁共享资源）。
792 0 : jettyHandle = 0;
793 0 : jfcHandle = 0;
794 0 : if (releaseCb_) {
795 0 : releaseCb_(endpointTag_);
796 0 : releaseCb_ = nullptr;
797 : }
798 0 : HCCL_INFO("[DevUbConnection][%s] EXTERNAL_INJECT mode, skip DestroyJetty, releaseCb invoked.", __func__);
799 : (void)ctxValid; // EXTERNAL_INJECT 模式无私有 JFC 需销毁，ctxValid 不影响
800 0 : }
801 :
802 112 : void DevUbConnection::ReleaseOwnedJettyAndJfc(bool ctxValid)
803 : {
804 112 : if (jettyHandle != 0) {
805 4 : if (!ctxValid) {
806 12 : HCCL_WARNING(
807 : "[DevUbConnection][%s] skip HrtRaUbDestroyJetty, "
808 : "rdmaHandle=%p invalid, jettyHandle=0x%llx",
809 : __func__, rdmaHandle, static_cast<unsigned long long>(jettyHandle));
810 : } else {
811 0 : HrtRaUbDestroyJetty(jettyHandle);
812 : }
813 4 : jettyHandle = 0;
814 : }
815 :
816 112 : if (engine_ == COMM_ENGINE_AIV && jfcHandle != 0) {
817 0 : if (!ctxValid) {
818 0 : HCCL_WARNING(
819 : "[DevUbConnection][%s] skip HrtRaUbDestroyJfc, "
820 : "rdmaHandle=%p invalid, jfcHandle=0x%llx",
821 : __func__, rdmaHandle, static_cast<unsigned long long>(jfcHandle));
822 : } else {
823 0 : HrtRaUbDestroyJfc(rdmaHandle, jfcHandle);
824 : }
825 0 : jfcHandle = 0;
826 : }
827 112 : }
828 :
829 112 : void DevUbConnection::ReleaseResource()
830 : {
831 112 : const bool ctxValid = (rdmaHandle != nullptr) && RdmaHandleManager::GetInstance().IsHandleValid(rdmaHandle);
832 :
833 112 : ReleaseRemoteJettyIfImported(ctxValid);
834 :
835 112 : if (releaseTpOnDestroy_) {
836 112 : ReleaseTp();
837 : }
838 :
839 : // EXTERNAL_INJECT 模式（主 connection）：jetty/JFC 由 JettyContext 统一管理，调 releaseCb_ 减引用
840 112 : if (jettyMode_ == JettyMode::EXTERNAL_INJECT) {
841 0 : ReleaseSharedJettyModeResources(ctxValid);
842 0 : return;
843 : }
844 :
845 : // SELF_CREATE + DetachJetty（临时 connection）：jetty/JFC 所有权已移交给 JettyContext，不销毁
846 112 : if (jettyDetached_) {
847 0 : HCCL_INFO("[DevUbConnection][%s] SELF_CREATE + DetachJetty, skip DestroyJetty/Jfc.", __func__);
848 0 : jettyHandle = 0;
849 0 : jfcHandle = 0;
850 0 : return;
851 : }
852 :
853 112 : ReleaseOwnedJettyAndJfc(ctxValid);
854 : }
855 :
856 1 : void DevUbConnection::CreateAivUrmaJfc()
857 : {
858 1 : jfcHandle = HrtRaUbCreateJfcUserCtl(rdmaHandle, cqInfo_);
859 3 : HCCL_INFO("[DevUbConnection][CreateAivUrmaJfc] create jfcHandle[%p] for rdmaHandle[%p].", jfcHandle, rdmaHandle);
860 1 : }
861 :
862 112 : DevUbConnection::~DevUbConnection() { DECTOR_TRY_CATCH("DevUbConnection", ReleaseResource()); }
863 :
864 : // Suspend接口当前已不使用，由框架调用触发析构流程
865 3 : bool DevUbConnection::Suspend()
866 : {
867 9 : HCCL_WARNING("[DevUbConnection][%s] should not be called.", __func__);
868 3 : if (status == RmaConnStatus::SUSPENDED) {
869 3 : HCCL_INFO("[DevUbConnection][%s] RmaConnStatus is SUSPENDED, status[%s].", __func__, status.Describe().c_str());
870 1 : return true;
871 : }
872 :
873 2 : if (status != RmaConnStatus::READY) {
874 2 : ThrowAbnormalStatus(std::string(__func__));
875 : }
876 :
877 1 : ReleaseResource();
878 1 : status = RmaConnStatus::SUSPENDED;
879 1 : return true;
880 : }
881 :
882 15 : static void PrepareUbSendWrReqParamForWriteOrRead(
883 : HrtRaUbSendWrReqParam& sendWrReq, const HrtUbSendWrOpCode sendWrOpCode, const MemoryBuffer& remoteMemBuf,
884 : const MemoryBuffer& localMemBuf, JettyHandle remoteJettyHandle, const SqeConfig& config, u32 cqeEnable = 1)
885 : {
886 15 : sendWrReq.cqeEn = cqeEnable;
887 15 : sendWrReq.opcode = sendWrOpCode;
888 15 : sendWrReq.size = localMemBuf.size;
889 15 : sendWrReq.localAddr = localMemBuf.addr;
890 15 : sendWrReq.remoteAddr = remoteMemBuf.addr;
891 :
892 15 : sendWrReq.lmemHandle = localMemBuf.memHandle;
893 15 : sendWrReq.rmemHandle = remoteMemBuf.memHandle;
894 15 : sendWrReq.handle = remoteJettyHandle;
895 :
896 : // 打印入参
897 45 : HCCL_INFO(
898 : "PrepareOneUbSendForRead params opCode=[%u], size=[%u], localAddr=[0x%llx], "
899 : "remoteAddr=[0x%llx], lmemHandle=[0x%llx], rmemHandle=[0x%llx], "
900 : "jettyHandle=[0x%llx], cqeEn=[%u], config=[%d]",
901 : static_cast<u32>(sendWrReq.opcode), sendWrReq.size, localMemBuf.addr, remoteMemBuf.addr, localMemBuf.memHandle,
902 : remoteMemBuf.memHandle, remoteJettyHandle, sendWrReq.cqeEn, config);
903 15 : }
904 :
905 9 : static void PrepareUbSendWrReqParamReduceInfo(HrtRaUbSendWrReqParam& sendWrReq, DataType dataType, ReduceOp reduceOp)
906 : {
907 9 : sendWrReq.inlineReduceFlag = true;
908 9 : sendWrReq.dataType = dataType;
909 9 : sendWrReq.reduceOp = reduceOp;
910 27 : HCCL_INFO(
911 : "PrepareUbSendWrReqParamReduceInfo params inlineReduceFlag[%u], dataType[%s], reduceOp[%s]",
912 : sendWrReq.inlineReduceFlag, dataType.Describe().c_str(), reduceOp.Describe().c_str());
913 9 : }
914 :
915 : static void
916 4 : PrepareUbSendWrReqParamNotifyInfo(HrtRaUbSendWrReqParam& sendWrReq, u64 data, const MemoryBuffer& remoteNotifyMemBuf)
917 : {
918 4 : sendWrReq.opcode = HrtUbSendWrOpCode::WRITE_WITH_NOTIFY;
919 4 : sendWrReq.notifyData = data;
920 4 : sendWrReq.notifyAddr = remoteNotifyMemBuf.addr;
921 4 : sendWrReq.notifyHandle = remoteNotifyMemBuf.memHandle;
922 12 : HCCL_INFO(
923 : "PrepareUbSendWrReqParamNotifyInfo params opCode[%u], "
924 : "notifyData[0x%llx], notifyAddr[0x%llx], notifyHandle[0x%llx]",
925 : static_cast<u32>(sendWrReq.opcode), sendWrReq.notifyData, sendWrReq.notifyAddr, sendWrReq.notifyHandle);
926 4 : }
927 :
928 : std::unique_ptr<BaseTask>
929 18 : DevUbConnection::ConstructTaskUbSend(const HrtRaUbSendWrRespParam& sendWrResp, const SqeConfig& config)
930 : {
931 18 : unique_ptr<BaseTask> result;
932 18 : if (opMode == OpMode::OPBASE) {
933 15 : if (config.wqeMode == WqeMode::DWQE) {
934 7 : result = make_unique<TaskUbDirectSend>(
935 7 : sendWrResp.funcId, sendWrResp.dieId, sendWrResp.jettyId, sendWrResp.dwqeSize, sendWrResp.dwqe);
936 8 : } else if (config.wqeMode == WqeMode::DB_SEND) {
937 : result
938 7 : = make_unique<TaskUbDbSend>(sendWrResp.jettyId, sendWrResp.funcId, sendWrResp.piVal, sendWrResp.dieId);
939 1 : } else if (config.wqeMode == WqeMode::WRITE_VALUE) {
940 3 : HCCL_INFO("[DevUbConnection::%s] dbAddr=[%llx], piVal=[%u]", __func__, dbAddr, sendWrResp.piVal);
941 1 : result = make_unique<TaskWriteValue>(dbAddr, sendWrResp.piVal);
942 : } else {
943 0 : auto msg = StringFormat("Invalid WqeMode[%s]", config.wqeMode.Describe().c_str());
944 0 : THROW<InvalidParamsException>(msg);
945 0 : }
946 3 : } else if (opMode == OpMode::OFFLOAD) {
947 8 : CHK_PRT_THROW(
948 : sendWrResp.piVal < piVal,
949 : HCCL_ERROR(
950 : "[DevUbConnection::%s] sendWrResp.piVal[%u] is less than piVal[%u]", __func__, sendWrResp.piVal, piVal),
951 : InvalidParamsException, "sendWrResp.piVal or piVal is invalid");
952 2 : u32 sendPiVal = sendWrResp.piVal - piVal;
953 2 : result = make_unique<TaskUbDbSend>(sendWrResp.jettyId, sendWrResp.funcId, sendPiVal, sendWrResp.dieId);
954 6 : HCCL_INFO(
955 : "[DevUbConnection::%s] sendPiVal[%u] piVal[%u] sendWrResp.piVal[%u]", __func__, sendPiVal, piVal,
956 : sendWrResp.piVal);
957 : } else {
958 0 : auto msg = StringFormat("Invalid OpMode[%s]", opMode.Describe().c_str());
959 0 : THROW<InvalidParamsException>(msg);
960 0 : }
961 :
962 17 : piVal = sendWrResp.piVal;
963 17 : return result;
964 1 : }
965 :
966 11 : void DevUbConnection::ProcessSlices(
967 : const MemoryBuffer& loc, const MemoryBuffer& rmt,
968 : std::function<void(const MemoryBuffer&, const MemoryBuffer&, u32)> processOneSlice, DataType dataType) const
969 : {
970 33 : HCCL_INFO("[DevUbConnection::%s] start", __func__);
971 :
972 : // reduce操作需要保证切片大小是数据类型大小的整数倍
973 11 : u32 sliceSize = UB_MAX_TRANS_SIZE;
974 11 : if (dataType != DataType::INVALID) {
975 7 : u32 dataTypeSize = DATA_TYPE_SIZE_MAP.at(dataType);
976 7 : sliceSize = UB_MAX_TRANS_SIZE / dataTypeSize * dataTypeSize;
977 : }
978 :
979 11 : u32 locBufSize = loc.size;
980 11 : u32 sliceNum = locBufSize / sliceSize;
981 11 : u32 lastSliceSize = locBufSize % sliceSize;
982 11 : u64 totalSize = static_cast<u64>(sliceNum) * static_cast<u64>(sliceSize);
983 11 : if (loc.addr > UINT64_MAX - totalSize || rmt.addr > UINT64_MAX - totalSize) {
984 0 : THROW<InternalException>("integer overflow occurs");
985 : }
986 11 : for (u32 sliceIdx = 0; sliceIdx < sliceNum; sliceIdx++) {
987 0 : MemoryBuffer locSlice(loc.addr + sliceIdx * sliceSize, sliceSize, loc.memHandle);
988 0 : MemoryBuffer rmtSlice(rmt.addr + sliceIdx * sliceSize, sliceSize, rmt.memHandle);
989 : // 当前是最后一片，且没有lastSlice时，启用cqe
990 0 : u32 cqeEnable = (sliceIdx == sliceNum - 1 && lastSliceSize == 0) ? 1 : 0;
991 0 : processOneSlice(locSlice, rmtSlice, cqeEnable);
992 : }
993 :
994 11 : if (lastSliceSize > 0) {
995 11 : MemoryBuffer lastLocSlice(loc.addr + sliceNum * sliceSize, lastSliceSize, loc.memHandle);
996 11 : MemoryBuffer lastRmtSlice(rmt.addr + sliceNum * sliceSize, lastSliceSize, rmt.memHandle);
997 11 : processOneSlice(lastLocSlice, lastRmtSlice, 1);
998 11 : sliceNum++;
999 : }
1000 :
1001 33 : HCCL_INFO(
1002 : "[DevUbConnection::%s] end, locBufSize[%u], sliceNum[%u], sliceSize[%u], lastSliceSize[%u]", __func__,
1003 : locBufSize, sliceNum, sliceSize, lastSliceSize);
1004 11 : }
1005 :
1006 4 : void DevUbConnection::ProcessSlicesWithNotify(
1007 : const MemoryBuffer& loc, const MemoryBuffer& rmt,
1008 : std::function<void(const MemoryBuffer&, const MemoryBuffer&, u32)> processOneSlice,
1009 : std::function<void(const MemoryBuffer&, const MemoryBuffer&)> processOneSliceWithNotify, DataType dataType) const
1010 : {
1011 12 : HCCL_INFO("[DevUbConnection::%s] start", __func__);
1012 :
1013 : // reduce操作需要保证切片大小是数据类型大小的整数倍
1014 4 : u32 sliceSize = UB_MAX_TRANS_SIZE;
1015 4 : if (dataType != DataType::INVALID) {
1016 2 : u32 dataTypeSize = DATA_TYPE_SIZE_MAP.at(dataType);
1017 2 : sliceSize = UB_MAX_TRANS_SIZE / dataTypeSize * dataTypeSize;
1018 : }
1019 :
1020 4 : u32 locBufSize = loc.size;
1021 4 : u32 sliceNum = locBufSize / sliceSize;
1022 4 : u32 lastSliceSize = locBufSize % sliceSize;
1023 4 : if (sliceNum > 0 && lastSliceSize == 0) {
1024 0 : sliceNum--;
1025 0 : lastSliceSize = sliceSize;
1026 : }
1027 :
1028 4 : for (u32 sliceIdx = 0; sliceIdx < sliceNum; sliceIdx++) {
1029 0 : MemoryBuffer locSlice(loc.addr + sliceIdx * sliceSize, sliceSize, loc.memHandle);
1030 0 : MemoryBuffer rmtSlice(rmt.addr + sliceIdx * sliceSize, sliceSize, rmt.memHandle);
1031 : // 固定会有lastSlice，则前面的cqe都不启用
1032 0 : processOneSlice(locSlice, rmtSlice, 0);
1033 : }
1034 :
1035 4 : if (lastSliceSize > 0) {
1036 4 : MemoryBuffer lastLocSlice(loc.addr + sliceNum * sliceSize, lastSliceSize, loc.memHandle);
1037 4 : MemoryBuffer lastRmtSlice(rmt.addr + sliceNum * sliceSize, lastSliceSize, rmt.memHandle);
1038 4 : processOneSliceWithNotify(lastLocSlice, lastRmtSlice);
1039 4 : sliceNum++;
1040 : }
1041 :
1042 12 : HCCL_INFO(
1043 : "[DevUbConnection::%s] end, locBufSize[%u], sliceNum[%u], sliceSize[%u], lastSliceSize[%u]", __func__,
1044 : locBufSize, sliceNum, sliceSize, lastSliceSize);
1045 4 : }
1046 :
1047 : unique_ptr<BaseTask>
1048 2 : DevUbConnection::PrepareRead(const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const SqeConfig& config)
1049 : {
1050 2 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "DevUbConnection::PrepareRead");
1051 :
1052 2 : if (localMemBuf.size == 0) {
1053 0 : return nullptr;
1054 : }
1055 :
1056 2 : HrtRaUbSendWrRespParam sendWrResp{};
1057 2 : ProcessSlices(
1058 2 : localMemBuf, remoteMemBuf, [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice, u32 cqeEnable) {
1059 2 : HrtRaUbSendWrReqParam sendWrReq = {};
1060 2 : PrepareUbSendWrReqParamForWriteOrRead(
1061 : sendWrReq, HrtUbSendWrOpCode::READ, rmtSlice, locSlice, remoteJettyHandle, config, cqeEnable);
1062 :
1063 2 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1064 2 : });
1065 :
1066 2 : return ConstructTaskUbSend(sendWrResp, config);
1067 : }
1068 :
1069 2 : unique_ptr<BaseTask> DevUbConnection::PrepareReadReduce(
1070 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType dataType, ReduceOp reduceOp,
1071 : const SqeConfig& config)
1072 : {
1073 2 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "DevUbConnection::PrepareReadReduce");
1074 :
1075 2 : if (localMemBuf.size == 0) {
1076 0 : return nullptr;
1077 : }
1078 :
1079 2 : HrtRaUbSendWrRespParam sendWrResp{};
1080 2 : ProcessSlices(
1081 : localMemBuf, remoteMemBuf,
1082 2 : [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice, u32 cqeEnable) {
1083 2 : HrtRaUbSendWrReqParam sendWrReq = {};
1084 2 : PrepareUbSendWrReqParamForWriteOrRead(
1085 : sendWrReq, HrtUbSendWrOpCode::READ, rmtSlice, locSlice, remoteJettyHandle, config, cqeEnable);
1086 2 : PrepareUbSendWrReqParamReduceInfo(sendWrReq, dataType, reduceOp);
1087 :
1088 2 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1089 2 : },
1090 : dataType);
1091 :
1092 2 : return ConstructTaskUbSend(sendWrResp, config);
1093 : }
1094 :
1095 6 : unique_ptr<BaseTask> DevUbConnection::PrepareWrite(
1096 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, const SqeConfig& config)
1097 : {
1098 8 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "DevUbConnection::PrepareWrite");
1099 :
1100 4 : if (localMemBuf.size == 0) {
1101 2 : return nullptr;
1102 : }
1103 :
1104 2 : HrtRaUbSendWrRespParam sendWrResp{};
1105 2 : ProcessSlices(
1106 2 : localMemBuf, remoteMemBuf, [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice, u32 cqeEnable) {
1107 2 : HrtRaUbSendWrReqParam sendWrReq = {};
1108 2 : PrepareUbSendWrReqParamForWriteOrRead(
1109 : sendWrReq, HrtUbSendWrOpCode::WRITE, rmtSlice, locSlice, remoteJettyHandle, config, cqeEnable);
1110 2 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1111 2 : });
1112 :
1113 2 : return ConstructTaskUbSend(sendWrResp, config);
1114 : }
1115 :
1116 7 : unique_ptr<BaseTask> DevUbConnection::PrepareWriteReduce(
1117 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType dataType, ReduceOp reduceOp,
1118 : const SqeConfig& config)
1119 : {
1120 8 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "DevUbConnection::PrepareWriteReduce");
1121 :
1122 6 : if (localMemBuf.size == 0) {
1123 1 : return nullptr;
1124 : }
1125 :
1126 5 : HrtRaUbSendWrRespParam sendWrResp{};
1127 5 : ProcessSlices(
1128 : localMemBuf, remoteMemBuf,
1129 5 : [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice, u32 cqeEnable) {
1130 5 : HrtRaUbSendWrReqParam sendWrReq = {};
1131 5 : PrepareUbSendWrReqParamForWriteOrRead(
1132 : sendWrReq, HrtUbSendWrOpCode::WRITE, rmtSlice, locSlice, remoteJettyHandle, config, cqeEnable);
1133 5 : PrepareUbSendWrReqParamReduceInfo(sendWrReq, dataType, reduceOp);
1134 5 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1135 5 : },
1136 : dataType);
1137 :
1138 5 : return ConstructTaskUbSend(sendWrResp, config);
1139 : }
1140 :
1141 : unique_ptr<BaseTask>
1142 3 : DevUbConnection::PrepareInlineWrite(const MemoryBuffer& remoteMemBuf, u64 data, const SqeConfig& config)
1143 : {
1144 3 : HrtRaUbSendWrReqParam sendWrReq = {};
1145 3 : sendWrReq.opcode = HrtUbSendWrOpCode::WRITE;
1146 3 : sendWrReq.remoteAddr = remoteMemBuf.addr;
1147 3 : sendWrReq.rmemHandle = remoteMemBuf.memHandle;
1148 3 : sendWrReq.handle = remoteJettyHandle;
1149 3 : sendWrReq.inlineFlag = true;
1150 3 : sendWrReq.inlineData = reinterpret_cast<u8*>(&data);
1151 3 : sendWrReq.size = sizeof(data);
1152 : /*
1153 : * 当前只有前后同步使用writeValue任务
1154 : * 由于writeValue任务不使能cqe，
1155 : * writeValue和dwqe混用会有潜在问题，所以后面需要区分开这两种任务模式
1156 : * 不在同一个connection里面既使用writeValue又使用dwqe
1157 : */
1158 3 : if (config.wqeMode == WqeMode::WRITE_VALUE && opMode == OpMode::OPBASE) {
1159 : // 当前只有inlineWrite使用write value
1160 : // 图模式不能使用writeValue
1161 : // writeValue 不需要使能cqe
1162 1 : sendWrReq.cqeEn = false;
1163 : }
1164 :
1165 9 : HCCL_INFO(
1166 : "DevUbConnection::PrepareInlineWrite params opCode=[%u], "
1167 : "remoteAddr=[0x%llx], rmemHandle=[0x%llx], remoteJettyHandle=[0x%llx], inlineFlag[%u], size=[%u], data=[%u]",
1168 : sendWrReq.opcode, sendWrReq.remoteAddr, sendWrReq.rmemHandle, sendWrReq.handle, sendWrReq.inlineFlag,
1169 : sendWrReq.size, static_cast<u32>(*sendWrReq.inlineData));
1170 3 : auto res = HrtRaUbPostSend(jettyHandle, sendWrReq);
1171 :
1172 6 : return ConstructTaskUbSend(res, config);
1173 : }
1174 :
1175 : inline HrtRaUbSendWrReqParam ConstructUbSendWrReqParamForWriteWithNotify(
1176 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, u64 data, const MemoryBuffer& remoteNotifyMemBuf)
1177 : {
1178 : HrtRaUbSendWrReqParam sendWrReq = {};
1179 : sendWrReq.opcode = HrtUbSendWrOpCode::WRITE_WITH_NOTIFY;
1180 : sendWrReq.size = remoteMemBuf.size;
1181 : sendWrReq.localAddr = localMemBuf.addr;
1182 : sendWrReq.remoteAddr = remoteMemBuf.addr;
1183 : sendWrReq.lmemHandle = localMemBuf.memHandle;
1184 : sendWrReq.rmemHandle = remoteMemBuf.memHandle;
1185 : sendWrReq.notifyData = data;
1186 : sendWrReq.notifyAddr = remoteNotifyMemBuf.addr;
1187 : sendWrReq.notifyHandle = remoteNotifyMemBuf.memHandle;
1188 :
1189 : return sendWrReq;
1190 : }
1191 :
1192 2 : unique_ptr<BaseTask> DevUbConnection::PrepareWriteWithNotify(
1193 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, u64 data, const MemoryBuffer& remoteNotifyMemBuf,
1194 : const SqeConfig& config)
1195 : {
1196 2 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "DevUbConnection::PrepareWriteWithNotify");
1197 :
1198 2 : if (localMemBuf.size == 0) {
1199 0 : return nullptr;
1200 : }
1201 :
1202 2 : HrtRaUbSendWrRespParam sendWrResp{};
1203 2 : ProcessSlicesWithNotify(
1204 : localMemBuf, remoteMemBuf,
1205 4 : [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice, u32 cqeEnable) {
1206 0 : HrtRaUbSendWrReqParam sendWrReq = {};
1207 0 : PrepareUbSendWrReqParamForWriteOrRead(
1208 : sendWrReq, HrtUbSendWrOpCode::WRITE, rmtSlice, locSlice, remoteJettyHandle, config, cqeEnable);
1209 0 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1210 0 : },
1211 2 : [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice) {
1212 2 : HrtRaUbSendWrReqParam sendWrReq = {};
1213 2 : PrepareUbSendWrReqParamForWriteOrRead(
1214 : sendWrReq, HrtUbSendWrOpCode::WRITE, rmtSlice, locSlice, remoteJettyHandle, config);
1215 2 : PrepareUbSendWrReqParamNotifyInfo(sendWrReq, data, remoteNotifyMemBuf);
1216 :
1217 2 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1218 2 : });
1219 :
1220 2 : return ConstructTaskUbSend(sendWrResp, config);
1221 : }
1222 :
1223 2 : unique_ptr<BaseTask> DevUbConnection::PrepareWriteReduceWithNotify(
1224 : const MemoryBuffer& remoteMemBuf, const MemoryBuffer& localMemBuf, DataType dataType, ReduceOp reduceOp, u64 data,
1225 : const MemoryBuffer& remoteNotifyMemBuf, const SqeConfig& config)
1226 : {
1227 2 : VerifySizeIsEqual(remoteMemBuf, localMemBuf, "DevUbConnection::PrepareWriteReduceWithNotify");
1228 :
1229 2 : if (localMemBuf.size == 0) {
1230 0 : return nullptr;
1231 : }
1232 :
1233 2 : HrtRaUbSendWrRespParam sendWrResp{};
1234 2 : ProcessSlicesWithNotify(
1235 : localMemBuf, remoteMemBuf,
1236 4 : [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice, u32 cqeEnable) {
1237 0 : HrtRaUbSendWrReqParam sendWrReq = {};
1238 0 : PrepareUbSendWrReqParamForWriteOrRead(
1239 : sendWrReq, HrtUbSendWrOpCode::WRITE, rmtSlice, locSlice, remoteJettyHandle, config, cqeEnable);
1240 0 : PrepareUbSendWrReqParamReduceInfo(sendWrReq, dataType, reduceOp);
1241 0 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1242 0 : },
1243 2 : [&](const MemoryBuffer& locSlice, const MemoryBuffer& rmtSlice) {
1244 2 : HrtRaUbSendWrReqParam sendWrReq = {};
1245 2 : PrepareUbSendWrReqParamForWriteOrRead(
1246 : sendWrReq, HrtUbSendWrOpCode::WRITE, rmtSlice, locSlice, remoteJettyHandle, config);
1247 2 : PrepareUbSendWrReqParamReduceInfo(sendWrReq, dataType, reduceOp);
1248 2 : PrepareUbSendWrReqParamNotifyInfo(sendWrReq, data, remoteNotifyMemBuf);
1249 2 : sendWrResp = HrtRaUbPostSend(jettyHandle, sendWrReq);
1250 2 : },
1251 : dataType);
1252 :
1253 2 : return ConstructTaskUbSend(sendWrResp, config);
1254 : }
1255 :
1256 152 : string DevUbConnection::Describe() const
1257 : {
1258 : return StringFormat(
1259 : "DevUbConnection[locAddr=%s, rmtAddr=%s, status=%s, dieId=%u, funcId=%u, jettyId=%u, sqBuffVa=%llx, "
1260 : "sqDepth=%u, maxReadSize=%u, maxWriteSize=%u, tpn=%u, dbAddr=0x%llx]",
1261 456 : locAddr.Describe().c_str(), rmtAddr.Describe().c_str(), status.Describe().c_str(), dieId, funcId, jettyId,
1262 608 : sqBuffVa, sqDepth, maxReadSize, maxWriteSize, tpn, dbAddr);
1263 : }
1264 :
1265 1 : HcclResult DevUbConnection::Describe(std::string& dfxMsg)
1266 : {
1267 1 : uint16_t udpSport = 0xFFFF; // 无法获取实际的udpSport，使用0xFFFF表示未知
1268 1 : if (tpProtocol == TpProtocol::TP) {
1269 1 : struct TpAttr tpAttr {};
1270 1 : uint32_t attrBitmap = 1 << 13; // 13对应dataUdpSrcport
1271 1 : TRY_CATCH_PRINT_ERROR(
1272 : u32 devicePhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1273 : HcclResult ret
1274 : = HrtRaGetTpAttrAsync(devicePhyId, rdmaHandle, tpInfo.tpHandle, attrBitmap, tpAttr, reqHandle);
1275 : if (ret == HCCL_E_NOT_SUPPORT) {
1276 : HCCL_ERROR(
1277 : "[DevUbConnection::%s] this package does not support RaGetTpAttrAsync for device,"
1278 : " please change new package, devPhyId[%u]",
1279 : __func__, devicePhyId);
1280 : return ret;
1281 : } else if (ret != HCCL_SUCCESS) {
1282 : HCCL_ERROR("[DevUbConnection::%s] failed, hccl result[%d]", __func__, ret);
1283 : return ret;
1284 : });
1285 1 : udpSport = tpAttr.dataUdpSrcport;
1286 : }
1287 1 : udpSport = udpSport & 0xFF;
1288 :
1289 : std::string dfxStr = StringFormat(
1290 : "chip id[%u] die id[%u] func id[%u] jetty id[%u] "
1291 : "local %s remote %s udp sport[%u]",
1292 1 : devLogicId, dieId, funcId, jettyId, locEid.Describe().c_str(), rmtEid.Describe().c_str(), udpSport);
1293 1 : dfxMsg += dfxStr;
1294 3 : HCCL_INFO("[DevUbConnection::%s] %s", __func__, dfxStr.c_str());
1295 1 : return HCCL_SUCCESS;
1296 1 : }
1297 :
1298 4 : void DevUbConnection::AddNop(const Stream& stream)
1299 : {
1300 4 : if (opMode != OpMode::OFFLOAD) {
1301 3 : HCCL_WARNING("[DevUbConnection][AddNop]Invalid OpMode[%s]", opMode.Describe().c_str());
1302 1 : return;
1303 : }
1304 3 : if (sqDepth < piVal) {
1305 1 : auto msg = StringFormat("Invalid piVal[%u], piVal should be less than or equal to sqDepth[%u]", piVal, sqDepth);
1306 1 : THROW<InvalidParamsException>(msg);
1307 1 : }
1308 2 : if (sqDepth == piVal) {
1309 0 : return;
1310 : }
1311 2 : u32 numNop = sqDepth - piVal;
1312 2 : HrtRaUbPostNops(jettyHandle, remoteJettyHandle, numNop);
1313 :
1314 : HrtUbDbInfo info;
1315 2 : info.dbNum = 1;
1316 2 : info.wrCqe = 0; // 默认值是0 不会cqe 如果传1，驱动分发，会给hccl cqe，用于维护ci指针。
1317 2 : info.info[0].functionId = funcId;
1318 2 : info.info[0].dieId = dieId;
1319 2 : info.info[0].jettyId = jettyId;
1320 2 : info.info[0].piValue = numNop;
1321 2 : HrtUbDbSend(info, stream.GetPtr());
1322 :
1323 2 : piVal = sqDepth;
1324 : }
1325 :
1326 4 : HrtUbJfcMode DevUbConnection::GetUbJfcMode() const { return jfcMode; }
1327 :
1328 5 : JettyHandle& DevUbConnection::GetJettyHandle() { return jettyHandle; }
1329 :
1330 2 : JettyHandle& DevUbConnection::GetRemoteJettyHandle() { return remoteJettyHandle; }
1331 :
1332 2 : RdmaHandle& DevUbConnection::GetRdmaHandle() { return rdmaHandle; }
1333 :
1334 6 : u32 DevUbConnection::GetPiVal() const { return piVal; }
1335 :
1336 5 : u32 DevUbConnection::GetCiVal() const { return ciVal; }
1337 :
1338 5 : u32 DevUbConnection::GetSqDepth() const { return sqDepth; }
1339 :
1340 2 : void DevUbConnection::UpdateCiVal(u32 ci) { ciVal = ci; }
1341 :
1342 7 : std::vector<DevUbConnection*> GetStarsPollUbConns(const std::vector<RmaConnection*>& rmaConns)
1343 : {
1344 7 : std::vector<DevUbConnection*> ubConns;
1345 8 : for (auto& rmaConn : rmaConns) {
1346 1 : if (rmaConn->GetRmaConnType() == RmaConnType::UB) {
1347 1 : if (dynamic_cast<DevUbConnection*>(rmaConn)->GetUbJfcMode() == HrtUbJfcMode::STARS_POLL) {
1348 1 : ubConns.emplace_back(dynamic_cast<DevUbConnection*>(rmaConn));
1349 : }
1350 : }
1351 : }
1352 7 : return ubConns;
1353 0 : }
1354 :
1355 7 : bool IfNeedUpdatingUbCi(const std::vector<DevUbConnection*>& ubConns)
1356 : {
1357 9 : for (auto& ubConn : ubConns) {
1358 2 : u32 pi = ubConn->GetPiVal();
1359 2 : u32 ci = ubConn->GetCiVal();
1360 2 : u32 sqDepth = ubConn->GetSqDepth();
1361 : // 考虑pi翻转场景
1362 2 : u32 extra = pi >= ci ? 0 : sqDepth;
1363 2 : constexpr u32 thresholdDivisor = 2;
1364 :
1365 2 : if (static_cast<double>(pi + extra - ci)
1366 2 : >= static_cast<double>(sqDepth) / thresholdDivisor) { // 当pi和ci差距大于sqDepth/2时，更新ci
1367 0 : return true;
1368 : }
1369 : }
1370 7 : return false;
1371 : }
1372 :
1373 0 : void DevUbConnection::SetMaxReadSize(u32 value) { maxReadSize = value; }
1374 :
1375 0 : void DevUbConnection::SetMaxWriteSize(u32 value) { maxWriteSize = value; }
1376 :
1377 : } // namespace Hccl
|
