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