Line data Source code
1 : /**
2 : * Copyright (c) 2026 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "host_ub_connection.h"
12 :
13 : #include <cstdlib>
14 :
15 : #include "exception_util.h"
16 : #include "rma_conn_exception.h"
17 : #include "rdma_handle_manager.h"
18 : #include "exchange_ub_conn_dto.h"
19 :
20 : namespace Hccl {
21 :
22 : constexpr u32 OPBASED_UB_SQ_DEPTH_MAX = 8192;
23 : constexpr u32 UB_SQ_OFFLOAD_DEPTH = 128;
24 : constexpr u32 UB_SQ_WQEBB_SIZE = 64;
25 : constexpr u32 UB_MAX_TRANS_SIZE = 256 * 1024 * 1024; // UB单次最大传输量256*1024*1024 Byte
26 : constexpr u32 WQE_NUM_PER_SQE = 4; // URMA约束每个SQE包含4个WQEBB
27 :
28 43 : HostUbConnection::HostUbConnection(const RdmaHandle rdmaHandle, const IpAddress &locAddr, const IpAddress &rmtAddr,
29 43 : const OpMode opMode, const HrtUbJfcMode jfcMode, u8 qos)
30 43 : : RmaConnection(nullptr, RmaConnType::UB), rdmaHandle(rdmaHandle), locAddr(locAddr), rmtAddr(rmtAddr),
31 43 : opMode(opMode), jfcMode(jfcMode), rmtEid(rmtAddr.GetReverseEid()), locEid(locAddr.GetReverseEid()), qos_(qos)
32 : {
33 43 : HCCL_INFO("[HostUbConnection::HostUbConnection] rmtEid=%s", rmtEid.Describe().c_str());
34 :
35 43 : auto dieIdAndFuncId = RdmaHandleManager::GetInstance().GetDieAndFuncId(rdmaHandle); // 获取dieId和FuncId
36 43 : dieId = dieIdAndFuncId.first;
37 43 : funcId = dieIdAndFuncId.second;
38 :
39 43 : jfcHandle = RdmaHandleManager::GetInstance().GetJfcHandle(rdmaHandle, cqInfo_, jfcMode);
40 :
41 43 : sqDepth = OPBASED_UB_SQ_DEPTH_MAX;
42 43 : if (opMode == OpMode::OFFLOAD) {
43 1 : sqDepth = UB_SQ_OFFLOAD_DEPTH;
44 : }
45 43 : HCCL_INFO("rdmaHandle[%p] locAddr[%s] rmtAddr[%s] opMode[%u] jfcMode[%s] dieId[%u] funcId[%u] jfcHandle[%llu] sqDepth[%u]",
46 : rdmaHandle, locAddr.Describe().c_str(), rmtAddr.Describe().c_str(), opMode, jfcMode.Describe().c_str(),
47 : dieId, funcId, jfcHandle, sqDepth);
48 43 : if (sqDepth > (UINT32_MAX / UB_SQ_WQEBB_SIZE / WQE_NUM_PER_SQE)) {
49 0 : THROW<InternalException>("integer overflow occurs");
50 : }
51 43 : }
52 :
53 1 : HostUbTpConnection::HostUbTpConnection(const RdmaHandle rdmaHandle, const IpAddress &locAddr, const IpAddress &rmtAddr,
54 1 : const OpMode opMode, const HrtUbJfcMode jfcMode, u8 qos)
55 1 : : HostUbConnection(rdmaHandle, locAddr, rmtAddr, opMode, jfcMode, qos)
56 : {
57 1 : tpProtocol = TpProtocol::TP;
58 1 : }
59 :
60 9 : HostUbCtpConnection::HostUbCtpConnection(const RdmaHandle rdmaHandle, const IpAddress &locAddr, const IpAddress &rmtAddr,
61 9 : const OpMode opMode, const HrtUbJfcMode jfcMode, u8 qos)
62 9 : : HostUbConnection(rdmaHandle, locAddr, rmtAddr, opMode, jfcMode, qos)
63 : {
64 9 : tpProtocol = TpProtocol::CTP;
65 9 : }
66 :
67 1 : std::vector<char> HostUbConnection::GetUniqueId() const
68 : {
69 1 : BinaryStream binaryStream;
70 1 : binaryStream << dieId;
71 1 : binaryStream << funcId;
72 1 : binaryStream << jettyId_;
73 :
74 1 : bool dwqeCacheLocked = false; // 待修改，该jetty是否支持dwqeCachedLocked，默认不支持
75 1 : u32 jfcPollMode = 0; // 待修改，0代表STARS POLL，1代表software Poll
76 1 : u64 sqCiAddr = 0; // 待修改，软件poll CQ情况下，需要AICPU从该地址中读取CI,依赖UB驱动支持
77 1 : std::vector<char> result;
78 1 : binaryStream << jfcPollMode;
79 1 : binaryStream << dwqeCacheLocked;
80 1 : binaryStream << dbAddr;
81 1 : binaryStream << sqCiAddr;
82 1 : binaryStream << sqBuffVa;
83 1 : binaryStream << sqDepth;
84 1 : binaryStream << tpn;
85 1 : binaryStream << rmtEid.raw;
86 1 : binaryStream << locEid.raw;
87 :
88 1 : binaryStream.Dump(result);
89 1 : HCCL_INFO("HostUbConnection::GetUniqueId:%s", Describe().c_str());
90 1 : HCCL_INFO("type=%s, jfcPollMode=%u, dwqeCacheLocked=%d, sqCiAddr=0x%llx", rmaConnType.Describe().c_str(),
91 : jfcPollMode, dwqeCacheLocked, sqCiAddr);
92 1 : return result;
93 1 : }
94 :
95 1 : void HostUbConnection::SetCqInfo(HcclAiRMACQ &cq)
96 : {
97 1 : cq.jfcId = cqInfo_.id;
98 1 : cq.cqVA = cqInfo_.va;
99 1 : cq.cqeSize = cqInfo_.cqeSize;
100 1 : cq.cqDepth = cqInfo_.cqDepth;
101 1 : cq.dbAddr = cqInfo_.swdbAddr;
102 1 : }
103 :
104 1 : void HostUbConnection::SetWqInfo(HcclAiRMAWQ &wq)
105 : {
106 1 : wq.jettyId = jettyId_;
107 1 : wq.dbAddr = dbAddr;
108 1 : wq.sqVA = sqBuffVa;
109 1 : wq.sqDepth = sqDepth * WQE_NUM_PER_SQE;
110 1 : wq.tp_id = tpn;
111 1 : errno_t ret = memcpy_s(wq.rmtEid, sizeof(wq.rmtEid), rmtEid.raw, sizeof(wq.rmtEid));
112 1 : if (ret != EOK) {
113 0 : HCCL_ERROR("[HostUbConnection][%s] memcpy_s failed, ret=%d", __func__, ret);
114 0 : ThrowAbnormalStatus(std::string(__func__));
115 : }
116 1 : }
117 :
118 1 : void HostUbConnection::Connect()
119 : {
120 1 : GetStatus();
121 1 : }
122 :
123 61 : inline uint32_t GetRandomNum()
124 : {
125 61 : uint32_t randNum = std::rand();
126 61 : return randNum;
127 : }
128 :
129 6 : RmaConnStatus HostUbConnection::GetStatus()
130 : {
131 6 : switch (ubConnStatus) {
132 2 : case UbConnStatus::INIT: {
133 2 : HCCL_INFO("[HostUbConnection][%s] start, status[%s], ubConnStatus[%s].", __func__, status.Describe().c_str(),
134 : ubConnStatus.Describe().c_str());
135 2 : if (!GetTpInfo()) {
136 2 : ubConnStatus = UbConnStatus::TP_INFO_GETTING;
137 2 : break;
138 : }
139 0 : CreateJetty();
140 0 : SetJettyInfo();
141 0 : ubConnStatus = UbConnStatus::JETTY_CREATED;
142 0 : status = RmaConnStatus::EXCHANGEABLE;
143 0 : break;
144 : }
145 1 : case UbConnStatus::TP_INFO_GETTING: {
146 1 : if (GetTpInfo()) {
147 1 : CreateJetty();
148 1 : SetJettyInfo();
149 1 : ubConnStatus = UbConnStatus::JETTY_CREATED;
150 1 : status = RmaConnStatus::EXCHANGEABLE;
151 : }
152 1 : break;
153 : }
154 0 : case UbConnStatus::JETTY_CREATED: {
155 0 : HCCL_INFO("[HostUbConnection][%s] status[%s] will not change, "
156 : "should call ImportRmtDto to change status.",
157 : __func__, status.Describe().c_str());
158 0 : break;
159 : }
160 1 : case UbConnStatus::JETTY_IMPORTING: {
161 1 : SetImportInfo();
162 1 : ubConnStatus = UbConnStatus::READY;
163 1 : status = RmaConnStatus::READY;
164 1 : break;
165 : }
166 1 : case UbConnStatus::READY:
167 1 : break;
168 1 : default:
169 2 : ThrowAbnormalStatus(std::string(__func__));
170 : }
171 :
172 5 : return status;
173 : }
174 :
175 3 : std::unique_ptr<Serializable> HostUbConnection::GetExchangeDto()
176 : {
177 3 : if (status != RmaConnStatus::READY && status != RmaConnStatus::EXCHANGEABLE) {
178 1 : HCCL_ERROR("[HostUbConnection][%s] status[%s] is not expected.", __func__,
179 : status.Describe().c_str());
180 2 : ThrowAbnormalStatus(std::string(__func__));
181 : }
182 :
183 2 : if (tpProtocol != TpProtocol::INVALID) {
184 2 : jettyImportCfg.localTpHandle = tpInfo.tpHandle;
185 :
186 2 : HCCL_INFO("[HostUbConnection][%s] tpEnable, localTpHandle[0x%llx] localPsn[%u].", __func__,
187 : jettyImportCfg.localTpHandle, jettyImportCfg.localPsn);
188 : }
189 :
190 : std::unique_ptr<ExchangeUbConnDto> dto
191 2 : = make_unique<ExchangeUbConnDto>(tokenValue, keySize, jettyImportCfg.localTpHandle, jettyImportCfg.localPsn);
192 2 : errno_t ret = memcpy_s(dto->qpKey, HRT_UB_QP_KEY_MAX_LEN, repJetty_.key, HRT_UB_QP_KEY_MAX_LEN);
193 2 : if (ret != EOK) {
194 0 : HCCL_ERROR("[HostUbConnection][%s] memcpy_s failed, ret=%d", __func__, ret);
195 0 : ThrowAbnormalStatus(std::string(__func__));
196 : }
197 4 : return std::unique_ptr<Serializable>(dto.release());
198 2 : }
199 :
200 2 : void HostUbConnection::ParseRmtExchangeDto(const Serializable &rmtDto)
201 : {
202 2 : auto dto = dynamic_cast<const ExchangeUbConnDto &>(rmtDto);
203 2 : HCCL_INFO("[HostUbConnection][%s] remoteConnDto[%s]", __func__, dto.Describe().c_str());
204 2 : remoteTokenValue = dto.tokenValue;
205 2 : errno_t ret = memcpy_s(remoteQpKey, HRT_UB_QP_KEY_MAX_LEN, dto.qpKey, HRT_UB_QP_KEY_MAX_LEN);
206 2 : if (ret != EOK) {
207 0 : HCCL_ERROR("[HostUbConnection][%s] memcpy_s failed, ret=%d", __func__, ret);
208 0 : ThrowAbnormalStatus(std::string(__func__));
209 : }
210 :
211 2 : if (tpProtocol != TpProtocol::INVALID) {
212 2 : jettyImportCfg.remoteTpHandle = dto.tpHandle;
213 2 : jettyImportCfg.remotePsn = dto.psn;
214 2 : HCCL_INFO("[HostUbConnection][%s] tpEnable, remoteTpHandle[0x%llx], remotePsn[%u].", __func__,
215 : jettyImportCfg.remoteTpHandle, jettyImportCfg.remotePsn);
216 : }
217 2 : }
218 :
219 3 : void HostUbConnection::ImportRmtDto()
220 : {
221 3 : if (ubConnStatus == UbConnStatus::READY) {
222 1 : HCCL_WARNING("[HostUbConnection][%s] import jetty already, %s.",
223 : __func__, Describe().c_str());
224 1 : return;
225 : }
226 :
227 2 : if (ubConnStatus != UbConnStatus::JETTY_CREATED) {
228 1 : HCCL_ERROR("[HostUbConnection][%s] failed, ubConnStatus[%s] is not expected.",
229 : __func__, ubConnStatus.Describe().c_str());
230 2 : ThrowAbnormalStatus(std::string(__func__));
231 : }
232 :
233 1 : ImportJetty();
234 1 : ubConnStatus = UbConnStatus::JETTY_IMPORTING;
235 : }
236 :
237 4 : void HostUbConnection::ThrowAbnormalStatus(std::string funcName)
238 : {
239 : auto errMsg = StringFormat("[HostUbConnection][%s] failed, [%s].",
240 4 : funcName.c_str(), Describe().c_str());
241 4 : status = RmaConnStatus::CONN_INVALID;
242 4 : ubConnStatus = UbConnStatus::CONN_INVALID;
243 4 : THROW<RmaConnException>(errMsg);
244 4 : }
245 :
246 0 : bool HostUbConnection::CheckRequestResult()
247 : {
248 0 : if (reqHandle == 0) {
249 0 : return true;
250 : }
251 :
252 0 : ReqHandleResult result = HrtRaGetAsyncReqResult(reqHandle);
253 0 : if (result == ReqHandleResult::NOT_COMPLETED) {
254 0 : return false;
255 : }
256 :
257 0 : if (result != ReqHandleResult::COMPLETED) {
258 0 : THROW<InternalException>("[HostUbConnection][%s] failed, result[%s] is unexpected.",
259 0 : __func__, result.Describe().c_str());
260 : }
261 :
262 0 : return true;
263 : }
264 :
265 1 : void HostUbConnection::CreateJetty()
266 : {
267 1 : if (sqDepth > UINT32_MAX / UB_SQ_WQEBB_SIZE / WQE_NUM_PER_SQE) {
268 0 : THROW<InternalException>("[HostUbConnection][%s] failed, sqDepth[%u] times "
269 : "UB_SQ_WQEBB_SIZE[%u] overflow uint32 max.", __func__, sqDepth, UB_SQ_WQEBB_SIZE);
270 : }
271 1 : u32 size = static_cast<u32>(sqDepth) * static_cast<u32>(UB_SQ_WQEBB_SIZE) * static_cast<u32>(WQE_NUM_PER_SQE);
272 1 : TokenIdHandle tokenIdHandle = RdmaHandleManager::GetInstance().GetTokenIdInfo(rdmaHandle).first;
273 : HrtRaUbCreateJettyParam req {
274 : jfcHandle, jfcHandle,
275 : GetUbToken(), tokenIdHandle,
276 : HrtJettyMode::STANDARD, // peer模式只支持JETTY_MODE_URMA_NORMAL
277 : 0, // HOST展开与AICPU展开传入jetty id为0，申请一个新的jetty
278 : 0, // va由底层分配，此处填0即可。
279 1 : size, 0, sqDepth}; // 非CCUv2不需要填写sqeBufIndex
280 1 : if (tpInfo.hasMappedJettyPriority) {
281 0 : req.qos = static_cast<u8>(tpInfo.mappedJettyPriority & 0xFU);
282 : }
283 1 : HCCL_INFO("[HostUbConnection][%s] jetty create qos[%u] (maps to attr.ub.priority lower 4 bits).", __func__,
284 : static_cast<unsigned int>(req.qos));
285 :
286 1 : repJetty_ = HrtRaUbCreateJetty(rdmaHandle, req);
287 1 : }
288 :
289 1 : void HostUbConnection::SetJettyInfo()
290 : {
291 1 : jettyId_ = repJetty_.id;
292 1 : jettyHandle_ = repJetty_.handle;
293 1 : jettyVa_ = repJetty_.jettyVa;
294 1 : sqBuffVa = repJetty_.sqBuffVa; // hccp提供
295 1 : HCCL_INFO("[HostUbConnection][%s] Get sqBuffVa is %llx.", __func__, sqBuffVa);
296 1 : keySize = repJetty_.keySize;
297 1 : dbAddr = repJetty_.dbVa;
298 1 : }
299 :
300 3 : bool HostUbConnection::GetTpInfo()
301 : {
302 3 : if (tpProtocol == TpProtocol::INVALID) { // 不感知tp建链，当前默认不支持
303 0 : HCCL_ERROR("[HostUbConnection][%s] failed, tpProtocol[%s] is not expected.",
304 : __func__, tpProtocol.Describe().c_str());
305 0 : ThrowAbnormalStatus(std::string(__func__));
306 : }
307 :
308 3 : int32_t devLogicId = HrtGetDevice();
309 3 : RaUbGetTpInfoParam p{};
310 3 : p.locAddr = locAddr;
311 3 : p.rmtAddr = rmtAddr;
312 3 : p.tpProtocol = tpProtocol;
313 3 : p.qos = static_cast<uint32_t>(qos_);
314 3 : auto ret = TpManager::GetInstance(devLogicId).GetTpInfo(p, tpInfo, true);
315 :
316 3 : switch (ret) {
317 2 : case HcclResult::HCCL_E_AGAIN:
318 2 : return false;
319 1 : case HcclResult::HCCL_SUCCESS:
320 1 : GenerateLocalPsn();
321 1 : return true;
322 0 : case HcclResult::HCCL_E_NOT_FOUND:
323 : default:
324 0 : HCCL_ERROR("[HostUbConnection][%s] failed, hccl result[%d]", __func__, ret);
325 0 : ThrowAbnormalStatus(std::string(__func__));
326 0 : break;
327 : }
328 0 : return true;
329 : }
330 :
331 1 : void HostUbConnection::GenerateLocalPsn()
332 : {
333 1 : jettyImportCfg.localPsn = GetRandomNum();
334 1 : }
335 :
336 1 : void HostUbConnection::ImportJetty()
337 : {
338 1 : HrtRaUbJettyImportedInParam in{};
339 1 : in.key = remoteQpKey;
340 1 : in.keyLen = keySize;
341 1 : in.tokenValue = remoteTokenValue;
342 1 : in.jettyImportCfg = jettyImportCfg;
343 1 : in.jettyImportCfg.protocol = tpProtocol;
344 :
345 1 : if (tpProtocol != TpProtocol::CTP && tpProtocol != TpProtocol::TP) {
346 0 : HCCL_ERROR("[HostUbConnection][%s] failed, tp protocol[%s] is not expected, %s.",
347 : __func__, tpProtocol.Describe().c_str(), Describe().c_str());
348 0 : ThrowAbnormalStatus(std::string(__func__));
349 : }
350 :
351 1 : remOutParam_ = RaUbTpImportJetty(rdmaHandle, in.key, in.keyLen, in.tokenValue, in.jettyImportCfg);
352 1 : }
353 :
354 2 : void HostUbConnection::SetImportInfo()
355 : {
356 2 : remoteJettyVa_ = remOutParam_.targetJettyVa;
357 2 : remoteJettyHandle_ = remOutParam_.handle;
358 2 : tpn = remOutParam_.tpn;
359 2 : return;
360 : }
361 :
362 45 : void HostUbConnection::ReleaseTp()
363 : {
364 45 : ReleaseUbConnectionTp(HrtGetDevice(), locAddr, rmtAddr, tpProtocol, tpInfo, static_cast<uint32_t>(qos_));
365 45 : }
366 :
367 44 : void HostUbConnection::ReleaseResource()
368 : {
369 44 : if (rdmaHandle && remoteJettyHandle_ != 0) {
370 2 : HrtRaUbUnimportJetty(rdmaHandle, remoteJettyHandle_);
371 2 : remoteJettyHandle_ = 0;
372 : }
373 :
374 44 : ReleaseTp();
375 :
376 44 : if (jettyHandle_ != 0) {
377 4 : HrtRaUbDestroyJetty(jettyHandle_);
378 4 : jettyHandle_ = 0;
379 : }
380 44 : }
381 :
382 43 : HostUbConnection::~HostUbConnection()
383 : {
384 43 : DECTOR_TRY_CATCH("HostUbConnection", ReleaseResource());
385 43 : }
386 :
387 : // Suspend接口当前已不使用，由框架调用触发析构流程
388 3 : bool HostUbConnection::Suspend()
389 : {
390 3 : HCCL_WARNING("[HostUbConnection][%s] should not be called.", __func__);
391 3 : return true;
392 : }
393 :
394 0 : std::unique_ptr<BaseTask> HostUbConnection::ConstructTaskUbSend(const HrtRaUbSendWrRespParam &sendWrResp,
395 : const SqeConfig &config) const
396 : {
397 : (void)sendWrResp;
398 : (void)config;
399 0 : unique_ptr<BaseTask> result;
400 0 : return result;
401 : }
402 :
403 0 : void HostUbConnection::ProcessSlices(const MemoryBuffer &loc, const MemoryBuffer &rmt,
404 : std::function<void(const MemoryBuffer &, const MemoryBuffer &, u32)> processOneSlice,
405 : DataType dataType) const
406 : {
407 : (void)loc;
408 : (void)rmt;
409 : (void)processOneSlice;
410 : (void)dataType;
411 0 : }
412 :
413 1 : unique_ptr<BaseTask> HostUbConnection::PrepareRead(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
414 : const SqeConfig &config)
415 : {
416 : (void)remoteMemBuf;
417 : (void)localMemBuf;
418 : (void)config;
419 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
420 1 : return nullptr;
421 : }
422 :
423 1 : unique_ptr<BaseTask> HostUbConnection::PrepareReadReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
424 : DataType dataType, ReduceOp reduceOp, const SqeConfig &config)
425 : {
426 : (void)remoteMemBuf;
427 : (void)localMemBuf;
428 : (void)dataType;
429 : (void)reduceOp;
430 : (void)config;
431 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
432 1 : return nullptr;
433 : }
434 :
435 1 : unique_ptr<BaseTask> HostUbConnection::PrepareWrite(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
436 : const SqeConfig &config)
437 : {
438 : (void)remoteMemBuf;
439 : (void)localMemBuf;
440 : (void)config;
441 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
442 1 : return nullptr;
443 : }
444 :
445 1 : unique_ptr<BaseTask> HostUbConnection::PrepareWriteReduce(const MemoryBuffer &remoteMemBuf,
446 : const MemoryBuffer &localMemBuf, DataType dataType,
447 : ReduceOp reduceOp, const SqeConfig &config)
448 : {
449 : (void)remoteMemBuf;
450 : (void)localMemBuf;
451 : (void)dataType;
452 : (void)reduceOp;
453 : (void)config;
454 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
455 1 : return nullptr;
456 : }
457 :
458 0 : unique_ptr<BaseTask> HostUbConnection::PrepareInlineWrite(const MemoryBuffer &remoteMemBuf, u64 data,
459 : const SqeConfig &config)
460 : {
461 : (void)remoteMemBuf;
462 : (void)data;
463 : (void)config;
464 0 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
465 0 : return nullptr;
466 : }
467 :
468 7 : string HostUbConnection::Describe() const
469 : {
470 : return StringFormat("HostUbConnection[locAddr=%s, rmtAddr=%s, status=%s, dieId=%u, funcId=%u, jettyId=%u, sqBuffVa=%llx, "
471 : "sqDepth=%u, tpn=%u, dbAddr=0x%llx]",
472 21 : locAddr.Describe().c_str(), rmtAddr.Describe().c_str(), status.Describe().c_str(), dieId,
473 28 : funcId, jettyId_, sqBuffVa, sqDepth, tpn, dbAddr);
474 : }
475 :
476 0 : void HostUbConnection::AddNop(const Stream &stream)
477 : {
478 : (void)stream;
479 0 : }
480 :
481 1 : HrtUbJfcMode HostUbConnection::GetUbJfcMode() const
482 : {
483 1 : return jfcMode;
484 : }
485 :
486 1 : JettyHandle& HostUbConnection::GetJettyHandle()
487 : {
488 1 : return jettyHandle_;
489 : }
490 :
491 1 : JettyHandle& HostUbConnection::GetRemoteJettyHandle()
492 : {
493 1 : return remoteJettyHandle_;
494 : }
495 :
496 1 : RdmaHandle& HostUbConnection::GetRdmaHandle()
497 : {
498 1 : return rdmaHandle;
499 : }
500 :
501 4 : u32 HostUbConnection::GetPiVal() const
502 : {
503 4 : return piVal;
504 : }
505 :
506 5 : u32 HostUbConnection::GetCiVal() const
507 : {
508 5 : return ciVal;
509 : }
510 :
511 5 : u32 HostUbConnection::GetSqDepth() const
512 : {
513 5 : return sqDepth;
514 : }
515 :
516 4 : uint64_t HostUbConnection::GetCqVa() const
517 : {
518 4 : return cqInfo_.va;
519 : }
520 :
521 1 : u64 HostUbConnection::GetJettyVa() const
522 : {
523 1 : return jettyVa_;
524 : }
525 :
526 1 : JettyHandle HostUbConnection::GetTJettyVa() const
527 : {
528 1 : return remoteJettyVa_;
529 : }
530 :
531 1 : void HostUbConnection::UpdateCiVal(u32 ci)
532 : {
533 1 : ciVal = ci;
534 1 : }
535 :
536 3 : bool IfNeedUpdatingUbCi(const std::vector<HostUbConnection *> &ubConns)
537 : {
538 5 : for (auto &ubConn : ubConns) {
539 3 : u32 pi = ubConn->GetPiVal();
540 3 : u32 ci = ubConn->GetCiVal();
541 3 : u32 sqDepth = ubConn->GetSqDepth();
542 : // 考虑pi翻转场景
543 3 : u32 extra = pi >= ci ? 0 : sqDepth;
544 3 : constexpr u32 thresholdDivisor = 2;
545 :
546 3 : if (static_cast<double>(pi + extra - ci) >= static_cast<double>(sqDepth) / thresholdDivisor) {
547 : // 当pi和ci差距大于sqDepth/2时，更新ci
548 1 : return true;
549 : }
550 : }
551 2 : return false;
552 : }
553 :
554 : } // namespace Hccl
|
