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