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