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