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