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