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 : HrtRaUbCreateJettyParam req {
273 : jfcHandle, jfcHandle,
274 : GetUbToken(), 0,
275 : HrtJettyMode::STANDARD, // peer模式只支持JETTY_MODE_URMA_NORMAL
276 : 0, // HOST展开与AICPU展开传入jetty id为0，申请一个新的jetty
277 : 0, // va由底层分配，此处填0即可。
278 1 : size, 0, sqDepth}; // 非CCUv2不需要填写sqeBufIndex
279 1 : if (tpInfo.hasMappedJettyPriority) {
280 0 : req.qos = static_cast<u8>(tpInfo.mappedJettyPriority & 0xFU);
281 : }
282 1 : HCCL_INFO("[HostUbConnection][%s] jetty create qos[%u] (maps to attr.ub.priority lower 4 bits).", __func__,
283 : static_cast<unsigned int>(req.qos));
284 :
285 1 : repJetty_ = HrtRaUbCreateJetty(rdmaHandle, req);
286 1 : }
287 :
288 1 : void HostUbConnection::SetJettyInfo()
289 : {
290 1 : jettyId_ = repJetty_.id;
291 1 : jettyHandle_ = repJetty_.handle;
292 1 : jettyVa_ = repJetty_.jettyVa;
293 1 : sqBuffVa = repJetty_.sqBuffVa; // hccp提供
294 1 : HCCL_INFO("[HostUbConnection][%s] Get sqBuffVa is %llx.", __func__, sqBuffVa);
295 1 : keySize = repJetty_.keySize;
296 1 : dbAddr = repJetty_.dbVa;
297 1 : }
298 :
299 3 : bool HostUbConnection::GetTpInfo()
300 : {
301 3 : if (tpProtocol == TpProtocol::INVALID) { // 不感知tp建链，当前默认不支持
302 0 : HCCL_ERROR("[HostUbConnection][%s] failed, tpProtocol[%s] is not expected.",
303 : __func__, tpProtocol.Describe().c_str());
304 0 : ThrowAbnormalStatus(std::string(__func__));
305 : }
306 :
307 3 : int32_t devLogicId = HrtGetDevice();
308 3 : RaUbGetTpInfoParam p{};
309 3 : p.locAddr = locAddr;
310 3 : p.rmtAddr = rmtAddr;
311 3 : p.tpProtocol = tpProtocol;
312 3 : p.qos = static_cast<uint32_t>(qos_);
313 3 : auto ret = TpManager::GetInstance(devLogicId).GetTpInfo(p, tpInfo, true);
314 :
315 3 : switch (ret) {
316 2 : case HcclResult::HCCL_E_AGAIN:
317 2 : return false;
318 1 : case HcclResult::HCCL_SUCCESS:
319 1 : GenerateLocalPsn();
320 1 : return true;
321 0 : case HcclResult::HCCL_E_NOT_FOUND:
322 : default:
323 0 : HCCL_ERROR("[HostUbConnection][%s] failed, hccl result[%d]", __func__, ret);
324 0 : ThrowAbnormalStatus(std::string(__func__));
325 0 : break;
326 : }
327 0 : return true;
328 : }
329 :
330 1 : void HostUbConnection::GenerateLocalPsn()
331 : {
332 1 : jettyImportCfg.localPsn = GetRandomNum();
333 1 : }
334 :
335 1 : void HostUbConnection::ImportJetty()
336 : {
337 1 : HrtRaUbJettyImportedInParam in{};
338 1 : in.key = remoteQpKey;
339 1 : in.keyLen = keySize;
340 1 : in.tokenValue = remoteTokenValue;
341 1 : in.jettyImportCfg = jettyImportCfg;
342 1 : in.jettyImportCfg.protocol = tpProtocol;
343 :
344 1 : if (tpProtocol != TpProtocol::CTP && tpProtocol != TpProtocol::TP) {
345 0 : HCCL_ERROR("[HostUbConnection][%s] failed, tp protocol[%s] is not expected, %s.",
346 : __func__, tpProtocol.Describe().c_str(), Describe().c_str());
347 0 : ThrowAbnormalStatus(std::string(__func__));
348 : }
349 :
350 1 : remOutParam_ = RaUbTpImportJetty(rdmaHandle, in.key, in.keyLen, in.tokenValue, in.jettyImportCfg);
351 1 : }
352 :
353 2 : void HostUbConnection::SetImportInfo()
354 : {
355 2 : remoteJettyVa_ = remOutParam_.targetJettyVa;
356 2 : remoteJettyHandle_ = remOutParam_.handle;
357 2 : tpn = remOutParam_.tpn;
358 2 : return;
359 : }
360 :
361 45 : void HostUbConnection::ReleaseTp()
362 : {
363 45 : ReleaseUbConnectionTp(HrtGetDevice(), locAddr, rmtAddr, tpProtocol, tpInfo, static_cast<uint32_t>(qos_));
364 45 : }
365 :
366 44 : void HostUbConnection::ReleaseResource()
367 : {
368 44 : if (rdmaHandle && remoteJettyHandle_ != 0) {
369 2 : HrtRaUbUnimportJetty(rdmaHandle, remoteJettyHandle_);
370 2 : remoteJettyHandle_ = 0;
371 : }
372 :
373 44 : ReleaseTp();
374 :
375 44 : if (jettyHandle_ != 0) {
376 4 : HrtRaUbDestroyJetty(jettyHandle_);
377 4 : jettyHandle_ = 0;
378 : }
379 44 : }
380 :
381 43 : HostUbConnection::~HostUbConnection()
382 : {
383 43 : DECTOR_TRY_CATCH("HostUbConnection", ReleaseResource());
384 43 : }
385 :
386 : // Suspend接口当前已不使用，由框架调用触发析构流程
387 3 : bool HostUbConnection::Suspend()
388 : {
389 3 : HCCL_WARNING("[HostUbConnection][%s] should not be called.", __func__);
390 3 : return true;
391 : }
392 :
393 0 : std::unique_ptr<BaseTask> HostUbConnection::ConstructTaskUbSend(const HrtRaUbSendWrRespParam &sendWrResp,
394 : const SqeConfig &config) const
395 : {
396 : (void)sendWrResp;
397 : (void)config;
398 0 : unique_ptr<BaseTask> result;
399 0 : return result;
400 : }
401 :
402 0 : void HostUbConnection::ProcessSlices(const MemoryBuffer &loc, const MemoryBuffer &rmt,
403 : std::function<void(const MemoryBuffer &, const MemoryBuffer &, u32)> processOneSlice,
404 : DataType dataType) const
405 : {
406 : (void)loc;
407 : (void)rmt;
408 : (void)processOneSlice;
409 : (void)dataType;
410 0 : }
411 :
412 1 : unique_ptr<BaseTask> HostUbConnection::PrepareRead(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
413 : const SqeConfig &config)
414 : {
415 : (void)remoteMemBuf;
416 : (void)localMemBuf;
417 : (void)config;
418 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
419 1 : return nullptr;
420 : }
421 :
422 1 : unique_ptr<BaseTask> HostUbConnection::PrepareReadReduce(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
423 : DataType dataType, ReduceOp reduceOp, const SqeConfig &config)
424 : {
425 : (void)remoteMemBuf;
426 : (void)localMemBuf;
427 : (void)dataType;
428 : (void)reduceOp;
429 : (void)config;
430 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
431 1 : return nullptr;
432 : }
433 :
434 1 : unique_ptr<BaseTask> HostUbConnection::PrepareWrite(const MemoryBuffer &remoteMemBuf, const MemoryBuffer &localMemBuf,
435 : const SqeConfig &config)
436 : {
437 : (void)remoteMemBuf;
438 : (void)localMemBuf;
439 : (void)config;
440 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
441 1 : return nullptr;
442 : }
443 :
444 1 : unique_ptr<BaseTask> HostUbConnection::PrepareWriteReduce(const MemoryBuffer &remoteMemBuf,
445 : const MemoryBuffer &localMemBuf, DataType dataType,
446 : ReduceOp reduceOp, const SqeConfig &config)
447 : {
448 : (void)remoteMemBuf;
449 : (void)localMemBuf;
450 : (void)dataType;
451 : (void)reduceOp;
452 : (void)config;
453 1 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
454 1 : return nullptr;
455 : }
456 :
457 0 : unique_ptr<BaseTask> HostUbConnection::PrepareInlineWrite(const MemoryBuffer &remoteMemBuf, u64 data,
458 : const SqeConfig &config)
459 : {
460 : (void)remoteMemBuf;
461 : (void)data;
462 : (void)config;
463 0 : HCCL_INFO("[HostUbConnection::%s] not supported yet.", __func__);
464 0 : return nullptr;
465 : }
466 :
467 7 : string HostUbConnection::Describe() const
468 : {
469 : return StringFormat("HostUbConnection[locAddr=%s, rmtAddr=%s, status=%s, dieId=%u, funcId=%u, jettyId=%u, sqBuffVa=%llx, "
470 : "sqDepth=%u, tpn=%u, dbAddr=0x%llx]",
471 21 : locAddr.Describe().c_str(), rmtAddr.Describe().c_str(), status.Describe().c_str(), dieId,
472 28 : funcId, jettyId_, sqBuffVa, sqDepth, tpn, dbAddr);
473 : }
474 :
475 0 : void HostUbConnection::AddNop(const Stream &stream)
476 : {
477 : (void)stream;
478 0 : }
479 :
480 1 : HrtUbJfcMode HostUbConnection::GetUbJfcMode() const
481 : {
482 1 : return jfcMode;
483 : }
484 :
485 1 : JettyHandle& HostUbConnection::GetJettyHandle()
486 : {
487 1 : return jettyHandle_;
488 : }
489 :
490 1 : JettyHandle& HostUbConnection::GetRemoteJettyHandle()
491 : {
492 1 : return remoteJettyHandle_;
493 : }
494 :
495 1 : RdmaHandle& HostUbConnection::GetRdmaHandle()
496 : {
497 1 : return rdmaHandle;
498 : }
499 :
500 4 : u32 HostUbConnection::GetPiVal() const
501 : {
502 4 : return piVal;
503 : }
504 :
505 5 : u32 HostUbConnection::GetCiVal() const
506 : {
507 5 : return ciVal;
508 : }
509 :
510 5 : u32 HostUbConnection::GetSqDepth() const
511 : {
512 5 : return sqDepth;
513 : }
514 :
515 4 : uint64_t HostUbConnection::GetCqVa() const
516 : {
517 4 : return cqInfo_.va;
518 : }
519 :
520 1 : u64 HostUbConnection::GetJettyVa() const
521 : {
522 1 : return jettyVa_;
523 : }
524 :
525 1 : JettyHandle HostUbConnection::GetTJettyVa() const
526 : {
527 1 : return remoteJettyVa_;
528 : }
529 :
530 1 : void HostUbConnection::UpdateCiVal(u32 ci)
531 : {
532 1 : ciVal = ci;
533 1 : }
534 :
535 3 : bool IfNeedUpdatingUbCi(const std::vector<HostUbConnection *> &ubConns)
536 : {
537 5 : for (auto &ubConn : ubConns) {
538 3 : u32 pi = ubConn->GetPiVal();
539 3 : u32 ci = ubConn->GetCiVal();
540 3 : u32 sqDepth = ubConn->GetSqDepth();
541 : // 考虑pi翻转场景
542 3 : u32 extra = pi >= ci ? 0 : sqDepth;
543 3 : constexpr u32 thresholdDivisor = 2;
544 :
545 3 : if (static_cast<double>(pi + extra - ci) >= static_cast<double>(sqDepth) / thresholdDivisor) {
546 : // 当pi和ci差距大于sqDepth/2时，更新ci
547 1 : return true;
548 : }
549 : }
550 2 : return false;
551 : }
552 :
553 : } // namespace Hccl
|
