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