Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "ccu_conn.h"
12 :
13 : #include <random>
14 : #include <sstream>
15 :
16 : #include "hcom_common.h"
17 : #include "exception_handler.h"
18 : #include "eid_info_mgr.h"
19 :
20 : #include "hccp_ctx.h"
21 :
22 : #include "rdma_handle_manager.h"
23 : #include "orion_adapter_hccp.h"
24 : #include "hcomm_res_mgr.h"
25 : #include "env_config/env_config_v2.h"
26 :
27 : namespace hcomm {
28 :
29 33 : CcuConnection::CcuConnection(
30 : const CommAddr& locAddr, const CommAddr& rmtAddr, const CcuChannelInfo& channelInfo,
31 33 : const std::vector<CcuJetty*>& ccuJettys, uint32_t qos)
32 33 : : locAddr_(locAddr),
33 33 : rmtAddr_(rmtAddr),
34 33 : channelInfo_(channelInfo),
35 33 : ccuJettys_(ccuJettys),
36 66 : qos_(qos)
37 33 : {}
38 :
39 7 : CcuRtpConnection::CcuRtpConnection(
40 : const CommAddr& locAddr, const CommAddr& rmtAddr, const CcuChannelInfo& channelInfo,
41 7 : const std::vector<CcuJetty*>& ccuJettys, uint32_t qos)
42 7 : : CcuConnection(locAddr, rmtAddr, channelInfo, ccuJettys, qos)
43 : {
44 7 : tpProtocol_ = TpProtocol::RTP;
45 7 : }
46 :
47 16 : CcuCtpConnection::CcuCtpConnection(
48 : const CommAddr& locAddr, const CommAddr& rmtAddr, const CcuChannelInfo& channelInfo,
49 16 : const std::vector<CcuJetty*>& ccuJettys, uint32_t qos)
50 16 : : CcuConnection(locAddr, rmtAddr, channelInfo, ccuJettys, qos)
51 : {
52 16 : tpProtocol_ = TpProtocol::CTP;
53 16 : }
54 :
55 15 : HcclResult CcuConnection::Init()
56 : {
57 15 : devLogicId_ = HcclGetThreadDeviceId();
58 15 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<uint32_t>(devLogicId_), devPhyId_));
59 :
60 : EXCEPTION_HANDLE_BEGIN
61 15 : auto& rdmaHandleMgr = Hccl::RdmaHandleManager::GetInstance();
62 15 : Hccl::IpAddress ipAddr{};
63 15 : CHK_RET(CommAddrToIpAddress(locAddr_, ipAddr));
64 15 : ctxHandle_ = rdmaHandleMgr.GetByIp(devPhyId_, ipAddr);
65 15 : CHK_PRT_RET(
66 : !rdmaHandleMgr.IsHandleValid(ctxHandle_),
67 : HCCL_ERROR(
68 : "[CcuConnection][%s] ctxHandle_[%p] is not valid, "
69 : "RdmaHandleManager may have DeInit this device",
70 : __func__, ctxHandle_),
71 : HcclResult::HCCL_E_INTERNAL);
72 :
73 15 : DevEidInfo eidInfo{};
74 15 : CHK_RET(EidInfoMgr::GetInstance(devPhyId_).GetEidInfoByAddr(locAddr_, eidInfo));
75 15 : dieId_ = static_cast<uint8_t>(eidInfo.dieId);
76 15 : funcId_ = eidInfo.funcId;
77 :
78 15 : EXCEPTION_HANDLE_END
79 :
80 15 : CHK_RET(GetLocalCcuRmaBufferInfo());
81 :
82 15 : jettyNum_ = channelInfo_.jettyInfos.size();
83 15 : CHK_PRT_RET(
84 : jettyNum_ == 0, HCCL_ERROR("[CcuConnection][%s] failed, jetty num[0] is unexpected.", __func__),
85 : HcclResult::HCCL_E_PARA);
86 :
87 15 : GenerateLocalPsn();
88 15 : status_ = CcuConnStatus::INIT;
89 15 : innerStatus_ = InnerStatus::INIT;
90 15 : return HcclResult::HCCL_SUCCESS;
91 : }
92 :
93 0 : CcuConnStatus CcuConnection::GetStatus()
94 : {
95 0 : if (status_ == CcuConnStatus::CONNECTED || status_ == CcuConnStatus::CONN_INVALID) {
96 0 : return status_;
97 : }
98 :
99 0 : if (StatusMachine() != HcclResult::HCCL_SUCCESS) {
100 0 : status_ = CcuConnStatus::CONN_INVALID;
101 0 : innerStatus_ = InnerStatus::CONN_INVALID;
102 : }
103 :
104 0 : return status_;
105 : }
106 :
107 15 : HcclResult CcuConnection::GetLocalCcuRmaBufferInfo()
108 : {
109 15 : uint64_t ccuBufSize = 0; // 暂未使用
110 15 : CHK_RET(CcuDevMgrImp::GetCcuResourceSpaceBufInfo(devLogicId_, dieId_, ccuBufAddr_, ccuBufSize));
111 :
112 15 : uint64_t tokenId = 0;
113 15 : uint64_t tokenValue = 0;
114 15 : CHK_RET(CcuDevMgrImp::GetCcuResourceSpaceTokenInfo(devLogicId_, dieId_, tokenId, tokenValue));
115 15 : ccuBufTokenId_ = static_cast<uint32_t>(tokenId);
116 15 : ccuBufTokenValue_ = static_cast<uint32_t>(tokenValue);
117 15 : return HcclResult::HCCL_SUCCESS;
118 : }
119 :
120 0 : HcclResult CcuConnection::StatusMachine()
121 : {
122 0 : if (status_ == CcuConnStatus::INIT) {
123 0 : CHK_RET(UpdateInitStatus());
124 0 : return HcclResult::HCCL_SUCCESS;
125 : }
126 :
127 0 : if (innerStatus_ == InnerStatus::JETTY_IMPORTING) {
128 0 : CHK_RET(UpdateExchangeStatus());
129 0 : return HcclResult::HCCL_SUCCESS;
130 : }
131 :
132 0 : return HcclResult::HCCL_SUCCESS;
133 : }
134 :
135 2 : HcclResult CcuConnection::GetTaTimeOut()
136 : {
137 2 : if (tpProtocol_ == TpProtocol::CTP) {
138 0 : uint32_t taTimeOutValue = 0;
139 0 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaCtpUbTimeOut(taTimeOutValue));
140 0 : errTimeout_ = static_cast<uint8_t>(taTimeOutValue);
141 0 : HCCL_INFO("[CcuConnection][%s] CTP, env errTimeout[%u].", __func__, errTimeout_);
142 0 : return HcclResult::HCCL_SUCCESS;
143 : }
144 :
145 2 : uint32_t taTimeOutValue = 0;
146 2 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaRtpUbTimeOut(taTimeOutValue));
147 2 : uint8_t envTaTimeOut = static_cast<uint8_t>(taTimeOutValue);
148 2 : uint32_t tpTimeOutMs = 0;
149 2 : (void)TpMgr::GetTpTotalTimeout(tpAttrInfo_, tpTimeOutMs);
150 2 : errTimeout_ = TpMgr::CalcTaTimeout(tpProtocol_, envTaTimeOut, tpTimeOutMs);
151 2 : return HcclResult::HCCL_SUCCESS;
152 : }
153 :
154 4 : HcclResult CcuConnection::UpdateInitStatus()
155 : {
156 4 : switch (innerStatus_) {
157 2 : case InnerStatus::INIT:
158 : case InnerStatus::TP_INFO_GETTING: {
159 2 : auto ret = GetTpInfo();
160 2 : if (ret == HcclResult::HCCL_E_AGAIN) {
161 0 : innerStatus_ = InnerStatus::TP_INFO_GETTING;
162 0 : return HcclResult::HCCL_SUCCESS;
163 : }
164 2 : CHK_RET(ret);
165 2 : CHK_PRT_RET(
166 : !tpInfo_.hasMappedJettyPriority,
167 : HCCL_ERROR("[CcuConnection][%s] TpMgr did not provide mappedJettyPriority.", __func__),
168 : HcclResult::HCCL_E_INTERNAL);
169 6 : for (auto* jetty : ccuJettys_) {
170 4 : CHK_RET(jetty->SetMappedJettyPriority(tpInfo_.mappedJettyPriority));
171 : }
172 2 : innerStatus_ = InnerStatus::TP_ATTR_GETTING;
173 2 : return HcclResult::HCCL_SUCCESS;
174 : }
175 1 : case InnerStatus::TP_ATTR_GETTING: {
176 1 : auto ret = GetTpAttr();
177 1 : if (ret == HcclResult::HCCL_E_AGAIN) {
178 0 : innerStatus_ = InnerStatus::TP_ATTR_GETTING;
179 0 : return HcclResult::HCCL_SUCCESS;
180 : }
181 1 : CHK_RET(ret);
182 :
183 1 : CHK_RET(GetTaTimeOut());
184 1 : innerStatus_ = InnerStatus::JETTY_CREATING;
185 1 : return HcclResult::HCCL_SUCCESS;
186 : }
187 1 : case InnerStatus::JETTY_CREATING: {
188 1 : auto ret = CreateJetty();
189 1 : if (ret == HcclResult::HCCL_E_AGAIN) {
190 0 : return HcclResult::HCCL_SUCCESS;
191 : }
192 1 : CHK_RET(ret);
193 1 : innerStatus_ = InnerStatus::EXCHANGEABLE;
194 1 : status_ = CcuConnStatus::EXCHANGEABLE;
195 1 : return HcclResult::HCCL_SUCCESS;
196 : }
197 0 : default:
198 0 : return ReturnErrorStatus(std::string(__func__));
199 : }
200 : }
201 :
202 1 : HcclResult CcuConnection::CreateJetty()
203 : {
204 1 : if (isJettyCreated_) {
205 0 : return HcclResult::HCCL_SUCCESS;
206 : }
207 :
208 1 : isJettyCreated_ = true;
209 1 : for (size_t i = 0; i < jettyNum_; i++) {
210 0 : auto ret = ccuJettys_[i]->CreateJetty(errTimeout_);
211 0 : if (ret == HcclResult::HCCL_E_AGAIN) {
212 : // 不提供日志避免刷屏
213 0 : isJettyCreated_ = isJettyCreated_ && false;
214 0 : continue;
215 : }
216 :
217 0 : if (ret != HcclResult::HCCL_SUCCESS) {
218 0 : isJettyCreated_ = true;
219 0 : HCCL_ERROR("[CcuConnection][%s] failed, hccl result[%d]", __func__, ret);
220 0 : return HcclResult::HCCL_E_NETWORK;
221 : }
222 : }
223 :
224 1 : return isJettyCreated_ ? HcclResult::HCCL_SUCCESS : HcclResult::HCCL_E_AGAIN;
225 : }
226 :
227 16 : inline uint32_t GetRandomNum()
228 : {
229 16 : uint32_t randNum = std::rand();
230 16 : return randNum;
231 : }
232 :
233 16 : void CcuConnection::GenerateLocalPsn() { jettyImportCfg_.localPsn = GetRandomNum(); }
234 :
235 7 : GetTpInfoParam CcuConnection::MakeGetTpInfoParam() const
236 : {
237 7 : GetTpInfoParam param;
238 7 : param.locAddr = locAddr_;
239 7 : param.rmtAddr = rmtAddr_;
240 7 : param.tpProtocol = tpProtocol_;
241 7 : param.qos = (qos_ > 7U) ? EnvConfig::UB_QOS_DEFAULT : (qos_ & 7U);
242 7 : param.slLevelCount = 0;
243 7 : param.loopFirstTpLowestSl = false;
244 7 : return param;
245 : }
246 :
247 2 : HcclResult CcuConnection::GetTpInfo()
248 : {
249 2 : if (tpProtocol_ == TpProtocol::INVALID) { // 不感知tp建链，当前默认不支持
250 0 : HCCL_ERROR(
251 : "[CcuConnection][%s] failed, tpProtocol[%s] is not expected.", __func__, tpProtocol_.Describe().c_str());
252 0 : return HcclResult::HCCL_E_PARA;
253 : }
254 :
255 2 : HcclResult ret = TpMgr::GetInstance(devPhyId_).GetTpInfo(MakeGetTpInfoParam(), tpInfo_);
256 2 : if (ret == HcclResult::HCCL_E_AGAIN) {
257 0 : return ret;
258 : }
259 :
260 2 : if (ret != HcclResult::HCCL_SUCCESS) {
261 0 : HCCL_ERROR("[CcuConnection][%s] failed, hccl result[%d]", __func__, ret);
262 0 : return HcclResult::HCCL_E_NETWORK;
263 : }
264 :
265 2 : jettyImportCfg_.localTpHandle = tpInfo_.tpHandle;
266 2 : return HcclResult::HCCL_SUCCESS;
267 : }
268 :
269 4 : HcclResult CcuConnection::GetTpAttr()
270 : {
271 4 : if (tpProtocol_ == TpProtocol::CTP) {
272 1 : HCCL_INFO("[CcuConnection][%s] CTP.", __func__);
273 1 : return HcclResult::HCCL_SUCCESS;
274 : }
275 :
276 3 : constexpr uint32_t kTpAttrRetryTimesInitBit = 0U;
277 3 : constexpr uint32_t kTpAttrAtBit = 1U;
278 3 : constexpr uint32_t TP_ATTR_BITMAP = (1U << kTpAttrRetryTimesInitBit) | (1U << kTpAttrAtBit);
279 : HcclResult ret
280 3 : = TpMgr::GetInstance(devPhyId_).GetTpAttr({tpInfo_.tpHandle, TP_ATTR_BITMAP}, tpAttrInfo_, ctxHandle_);
281 3 : if (ret == HcclResult::HCCL_E_AGAIN) {
282 1 : return ret;
283 : }
284 :
285 2 : if (ret != HcclResult::HCCL_SUCCESS) {
286 0 : HCCL_ERROR("[CcuConnection][%s] failed, hccl result[%d]", __func__, ret);
287 0 : return HcclResult::HCCL_E_NETWORK;
288 : }
289 :
290 2 : return HcclResult::HCCL_SUCCESS;
291 : }
292 :
293 0 : HcclResult CcuConnection::Serialize(std::vector<char>& dtoData)
294 : {
295 0 : if (status_ != CcuConnStatus::EXCHANGEABLE) {
296 0 : HCCL_ERROR(
297 : "[CcuConnection][%s] failed, not init completed yet, "
298 : "status[%s].",
299 : __func__, status_.Describe().c_str());
300 0 : return HcclResult::HCCL_E_INTERNAL;
301 : }
302 :
303 0 : Hccl::BinaryStream dtoStream;
304 0 : dtoStream << ccuBufAddr_;
305 0 : dtoStream << ccuBufTokenId_;
306 0 : dtoStream << ccuBufTokenValue_;
307 0 : HCCL_INFO("[CcuConnection][%s], ccuBufAddr[%llx]", __func__, ccuBufAddr_);
308 :
309 0 : dtoStream << jettyNum_;
310 0 : HCCL_INFO("[CcuConnection][%s], jettyNum[%u]", __func__, jettyNum_);
311 0 : for (const auto& ccuJetty : ccuJettys_) {
312 0 : dtoStream << ccuJetty->GetCreateJettyParam().tokenValue;
313 0 : const auto& outParam = ccuJetty->GetJettyedOutParam();
314 0 : dtoStream << outParam.key;
315 0 : dtoStream << outParam.keySize;
316 : }
317 :
318 0 : if (tpProtocol_ != TpProtocol::INVALID) {
319 0 : dtoStream << jettyImportCfg_.localTpHandle;
320 0 : dtoStream << jettyImportCfg_.localPsn;
321 0 : HCCL_INFO(
322 : "[CcuConnection][%s] tpProtocol[%s], localTpHandle[0x%llx], localPsn[%u].", __func__,
323 : tpProtocol_.Describe().c_str(), jettyImportCfg_.localTpHandle, jettyImportCfg_.localPsn);
324 : }
325 :
326 0 : dtoData.clear();
327 0 : dtoStream.Dump(dtoData);
328 0 : return HcclResult::HCCL_SUCCESS;
329 0 : }
330 :
331 0 : HcclResult CcuConnection::Deserialize(const std::vector<char>& dtoData)
332 : {
333 0 : if (status_ != CcuConnStatus::EXCHANGEABLE) {
334 0 : HCCL_ERROR(
335 : "[CcuConnection][%s] failed, not init completed yet, "
336 : "status[%s].",
337 : __func__, status_.Describe().c_str());
338 0 : return HcclResult::HCCL_E_INTERNAL;
339 : }
340 :
341 0 : std::vector<char> rmtDtoData = dtoData;
342 0 : Hccl::BinaryStream dtoStream(rmtDtoData);
343 0 : dtoStream >> rmtCcuBufAddr_;
344 0 : dtoStream >> rmtCcuBufTokenId_;
345 0 : dtoStream >> rmtCcuBufTokenValue_;
346 0 : HCCL_INFO("[CcuConnection][%s], rmtCcuBufAddr[%llx].", __func__, rmtCcuBufAddr_);
347 :
348 0 : uint32_t remoteJettySize{0};
349 0 : dtoStream >> remoteJettySize;
350 :
351 0 : importJettyCtxs_.clear();
352 0 : importJettyCtxs_.resize(remoteJettySize);
353 0 : HCCL_INFO("[CcuConnection][%s], remoteJettySize[%u].", __func__, remoteJettySize);
354 :
355 0 : for (auto& importCtx : importJettyCtxs_) {
356 0 : dtoStream >> importCtx.inParam.tokenValue;
357 0 : dtoStream >> importCtx.remoteQpKey; // 保存key数组
358 0 : importCtx.inParam.key = importCtx.remoteQpKey; // 保存指针用于接口调用
359 0 : dtoStream >> importCtx.inParam.keyLen;
360 : }
361 :
362 0 : if (tpProtocol_ != TpProtocol::INVALID) {
363 0 : dtoStream >> jettyImportCfg_.remoteTpHandle;
364 0 : dtoStream >> jettyImportCfg_.remotePsn;
365 :
366 0 : HCCL_INFO(
367 : "[CcuConnection][%s] tpEnable, remoteTpHandle[0x%llx], remotePsn[%u].", __func__,
368 : jettyImportCfg_.remoteTpHandle, jettyImportCfg_.remotePsn);
369 : }
370 :
371 0 : return HcclResult::HCCL_SUCCESS;
372 0 : }
373 :
374 0 : HcclResult CcuConnection::ImportJetty()
375 : {
376 0 : if (isJettyImported_) {
377 0 : HCCL_INFO("[CcuConnection][%s] taJettys has been imported already.", __func__);
378 0 : return HcclResult::HCCL_SUCCESS;
379 : }
380 :
381 0 : if (innerStatus_ != InnerStatus::EXCHANGEABLE) {
382 0 : return ReturnErrorStatus(std::string(__func__));
383 : }
384 :
385 : // importJettyCtxs_.resize(jettyNum_);
386 0 : if (jettyNum_ != importJettyCtxs_.size()) {
387 0 : HCCL_ERROR(
388 : "[CcuConnection][%s] failed to ImportJetty, "
389 : "jettyNum[%u] is not equal to importJettyCtxs.size[%u].",
390 : __func__, jettyNum_, importJettyCtxs_.size());
391 0 : return ReturnErrorStatus(std::string(__func__));
392 : }
393 :
394 0 : ResetRequestCtxs();
395 0 : for (size_t i = 0; i < jettyNum_; i++) {
396 0 : if (StartImportJettyRequest(i, reqHandles_[i]) != HcclResult::HCCL_SUCCESS) {
397 0 : return ReturnErrorStatus(std::string(__func__));
398 : }
399 : }
400 :
401 0 : innerStatus_ = InnerStatus::JETTY_IMPORTING;
402 0 : return HcclResult::HCCL_SUCCESS;
403 : }
404 :
405 0 : void CcuConnection::ResetRequestCtxs()
406 : {
407 0 : reqHandles_.clear();
408 0 : reqHandles_.resize(jettyNum_);
409 :
410 0 : reqDataBuffers_.clear();
411 0 : reqDataBuffers_.resize(jettyNum_);
412 :
413 0 : remoteJettyHandlePtrs_.clear();
414 0 : remoteJettyHandlePtrs_.resize(jettyNum_);
415 0 : }
416 :
417 0 : HcclResult CcuConnection::StartImportJettyRequest(uint32_t jettyIndex, RequestHandle& reqHandle)
418 : {
419 0 : if (tpProtocol_ == TpProtocol::INVALID) {
420 0 : return ReturnErrorStatus(std::string(__func__));
421 : }
422 :
423 0 : auto& importCtx = importJettyCtxs_[jettyIndex];
424 0 : auto& importCtxInParam = importCtx.inParam;
425 0 : importCtxInParam.jettyImportCfg = jettyImportCfg_;
426 0 : importCtxInParam.jettyImportCfg.protocol = tpProtocol_;
427 0 : CHK_RET(HccpUbTpImportJettyAsync(
428 : ctxHandle_, importCtxInParam, reqDataBuffers_[jettyIndex], remoteJettyHandlePtrs_[jettyIndex], reqHandle));
429 :
430 0 : return HcclResult::HCCL_SUCCESS;
431 : }
432 :
433 0 : HcclResult CcuConnection::CheckRequestResults()
434 : {
435 0 : if (reqHandles_.size() == 0) {
436 0 : return HcclResult::HCCL_SUCCESS;
437 : }
438 :
439 : // 检查所有下发异步请求是否完成
440 0 : std::vector<size_t> completedReqs;
441 0 : const uint32_t reqSize = reqHandles_.size();
442 0 : for (size_t i = 0; i < reqSize; i++) {
443 0 : RequestResult result = HccpGetAsyncReqResult(reqHandles_[i]);
444 0 : if (result == RequestResult::NOT_COMPLETED) {
445 0 : continue;
446 : }
447 :
448 0 : if (result != RequestResult::COMPLETED) {
449 0 : HCCL_ERROR("[CcuConnection][%s] failed, result[%s] is unexpected.", __func__, result.Describe().c_str());
450 0 : return HcclResult::HCCL_E_NETWORK;
451 : }
452 :
453 : // 记录已完成的reqHandles
454 0 : completedReqs.push_back(i);
455 : }
456 :
457 : // 删除已完成的reqHandles，避免重复查询
458 0 : for (int i = completedReqs.size() - 1; i >= 0; --i) {
459 0 : reqHandles_.erase(reqHandles_.begin() + completedReqs[i]);
460 : }
461 :
462 : // 检查是否有剩余reqHandles
463 0 : return reqHandles_.size() == 0 ? HcclResult::HCCL_SUCCESS : HcclResult::HCCL_E_AGAIN;
464 0 : }
465 :
466 0 : HcclResult CcuConnection::UpdateExchangeStatus()
467 : {
468 : // 状态机保证为 InnerStatus::JETTY_IMPORTING
469 0 : auto ret = CheckRequestResults();
470 0 : if (ret == HcclResult::HCCL_E_AGAIN) {
471 0 : return HcclResult::HCCL_SUCCESS; // 操作成功，保持当前状态
472 : }
473 0 : CHK_RET(ret);
474 :
475 0 : for (size_t i = 0; i < jettyNum_; i++) {
476 0 : auto& outParam = importJettyCtxs_[i].outParam;
477 0 : struct QpImportInfoT* infoPtr = reinterpret_cast<QpImportInfoT*>(reqDataBuffers_[i].data());
478 0 : outParam.handle = reinterpret_cast<TargetJettyHandle>(remoteJettyHandlePtrs_[i]);
479 0 : outParam.targetJettyVa = infoPtr->out.ub.tjettyHandle; // 该信息当前未使用
480 0 : outParam.tpn = infoPtr->out.ub.tpn;
481 : }
482 0 : isJettyImported_ = true;
483 :
484 0 : CHK_RET(ConfigChannel());
485 0 : status_ = CcuConnStatus::CONNECTED;
486 0 : innerStatus_ = InnerStatus::CONNECTED;
487 0 : return HcclResult::HCCL_SUCCESS;
488 : }
489 :
490 0 : HcclResult CcuConnection::ConfigChannel()
491 : {
492 0 : if (jettyNum_ != importJettyCtxs_.size()) {
493 0 : HCCL_ERROR(
494 : "[CcuConnection][%s] failed, jettyNum[%u] is not equal to "
495 : "importJettyCtxs.size[%u].",
496 : __func__, jettyNum_, importJettyCtxs_.size());
497 0 : return HcclResult::HCCL_E_INTERNAL;
498 : }
499 :
500 0 : ChannelCfg cfg{};
501 0 : cfg.channelId = channelInfo_.channelId;
502 0 : Hccl::IpAddress rmtAddr{};
503 0 : CHK_RET(CommAddrToIpAddress(rmtAddr_, rmtAddr));
504 0 : CHK_RET(IpAddressToReverseHcclEid(rmtAddr, cfg.remoteEid)); // 配置ccu硬件需要使用反向eid
505 0 : cfg.tpn = importJettyCtxs_[0].outParam.tpn; // tp handle复用所以tpn一致
506 0 : cfg.remoteCcuVa = rmtCcuBufAddr_;
507 0 : cfg.memTokenId = rmtCcuBufTokenId_;
508 0 : cfg.memTokenValue = rmtCcuBufTokenValue_;
509 :
510 0 : for (size_t i = 0; i < jettyNum_; i++) {
511 0 : const auto& ccuJetty = ccuJettys_[i];
512 0 : const auto& inParam = ccuJetty->GetCreateJettyParam();
513 0 : const auto& outParam = ccuJetty->GetJettyedOutParam();
514 0 : const auto& jettyInfo = channelInfo_.jettyInfos[i];
515 0 : cfg.jettyCfgs.emplace_back(JettyCfg{
516 0 : jettyInfo.jettyCtxId, outParam.dbVa, outParam.dbTokenId,
517 0 : inParam.tokenValue}); // 安全问题，禁止打印token相关信息
518 : }
519 :
520 0 : CHK_RET(CcuDevMgrImp::ConfigChannel(devLogicId_, dieId_, cfg));
521 0 : return HcclResult::HCCL_SUCCESS;
522 0 : }
523 :
524 33 : CcuConnection::~CcuConnection() { (void)ReleaseConnRes(); }
525 :
526 34 : HcclResult CcuConnection::ReleaseConnRes()
527 : {
528 34 : const bool ctxValid = ctxHandle_ != nullptr && Hccl::RdmaHandleManager::GetInstance().IsHandleValid(ctxHandle_);
529 :
530 34 : for (auto& item : importJettyCtxs_) {
531 0 : if (item.outParam.handle != 0) {
532 0 : if (!ctxValid) {
533 0 : HCCL_WARNING(
534 : "[CcuConnection][%s] skip RaCtxQpUnimport, ctxHandle=%p invalid, "
535 : "remoteJettyHandle=%p",
536 : __func__, ctxHandle_, item.outParam.handle);
537 0 : item.outParam.handle = 0;
538 0 : continue;
539 : }
540 0 : int32_t ret = RaCtxQpUnimport(ctxHandle_, item.outParam.handle);
541 0 : item.outParam.handle = 0;
542 0 : if (ret != 0) {
543 0 : HCCL_ERROR(
544 : "[CcuComponent][%s] failed but passed, ctxHandle[%p] "
545 : "remoteJettyHandle[%p], devLogicId[%d].",
546 : __func__, ctxHandle_, item.outParam.handle, devLogicId_);
547 0 : status_ = CcuConnStatus::CONN_INVALID;
548 0 : innerStatus_ = InnerStatus::CONN_INVALID;
549 : }
550 : }
551 : }
552 34 : importJettyCtxs_.clear();
553 :
554 34 : if (tpProtocol_ == TpProtocol::RTP && tpInfo_.tpHandle != 0) {
555 5 : (void)TpMgr::GetInstance(devPhyId_).ReleaseTpAttr(tpInfo_.tpHandle, tpAttrInfo_);
556 : }
557 :
558 34 : if (tpInfo_.tpHandle != 0) { // tp handle 复用，只释放一次
559 5 : (void)TpMgr::GetInstance(devPhyId_).ReleaseTpInfo(MakeGetTpInfoParam(), tpInfo_);
560 5 : tpInfo_.tpHandle = 0;
561 5 : tpInfo_.hasMappedJettyPriority = false;
562 : }
563 : // CcuJetty 生命周期跟随通信域CcuJettyMgr
564 : // 不需要connection主动销毁
565 34 : return HcclResult::HCCL_SUCCESS;
566 : }
567 :
568 0 : HcclResult CcuConnection::ReturnErrorStatus(const std::string& funcName)
569 : {
570 0 : std::string errMsg = Hccl::StringFormat("[CcuConnection][%s] failed, [%s].", funcName.c_str(), Describe().c_str());
571 0 : status_ = CcuConnStatus::CONN_INVALID;
572 0 : innerStatus_ = InnerStatus::CONN_INVALID;
573 0 : HCCL_ERROR("%s", errMsg.c_str());
574 0 : return HcclResult::HCCL_E_INTERNAL;
575 0 : }
576 :
577 0 : std::string CcuConnection::Describe()
578 : {
579 0 : Hccl::IpAddress locAddr{}, rmtAddr{};
580 0 : (void)CommAddrToIpAddress(locAddr_, locAddr);
581 0 : (void)CommAddrToIpAddress(rmtAddr_, rmtAddr);
582 : return Hccl::StringFormat(
583 : "[CcuConnection[locAddr=%s, rmtAddr=%s, protocol=%s, "
584 : "status=%s, innerStatus=%s, [dieId=%u, channelId=%u, jettyNum=%u]]]",
585 0 : locAddr.Describe().c_str(), rmtAddr.Describe().c_str(), tpProtocol_.Describe().c_str(),
586 0 : status_.Describe().c_str(), innerStatus_.Describe().c_str(), dieId_, channelInfo_.channelId, jettyNum_);
587 : }
588 :
589 4 : HcclResult CcuConnection::Describe(std::string& dfxMsg)
590 : {
591 4 : uint16_t udpSport = 0xFFFF; // 无法获取实际的udpSport，使用0xFFFF表示未知
592 4 : if (tpProtocol_ == TpProtocol::RTP) {
593 4 : struct TpAttr tpAttr {};
594 4 : uint32_t attrBitmap = 1 << 13; // 13对应dataUdpSrcport
595 : EXCEPTION_HANDLE_BEGIN
596 : // HrtRaGetTpAttrAsync：封装内已同步等待，返回时 tpAttr 已就绪
597 : HcclResult ret
598 4 : = Hccl::HrtRaGetTpAttrAsync(devPhyId_, ctxHandle_, tpInfo_.tpHandle, attrBitmap, tpAttr, reqHandles_[0]);
599 4 : if (ret == HCCL_E_NOT_SUPPORT) {
600 1 : HCCL_ERROR(
601 : "[DevUbConnection::%s] failed, this package does not support RaGetTpAttrAsync for device,"
602 : " please change new package. devPhyId[%u]",
603 : __func__, devPhyId_);
604 2 : return ret;
605 3 : } else if (ret != HCCL_SUCCESS) {
606 1 : HCCL_ERROR("[DevUbConnection::%s] failed, hccl result[%d]", __func__, ret);
607 1 : return ret;
608 : }
609 0 : EXCEPTION_HANDLE_END
610 2 : udpSport = tpAttr.dataUdpSrcport;
611 : }
612 2 : udpSport = udpSport & 0xFF;
613 :
614 2 : std::ostringstream oss;
615 6 : for (size_t i = 0; i < ccuJettys_.size(); ++i) {
616 4 : uint16_t jettyId = ccuJettys_[i]->GetJettyedOutParam().id;
617 4 : if (i != 0) {
618 2 : oss << ", ";
619 : }
620 4 : oss << jettyId;
621 : }
622 2 : std::string jettyIds = oss.str();
623 :
624 2 : Hccl::IpAddress locAddr{}, rmtAddr{};
625 2 : CHK_RET(CommAddrToIpAddress(locAddr_, locAddr));
626 2 : CHK_RET(CommAddrToIpAddress(rmtAddr_, rmtAddr));
627 2 : Hccl::Eid locEid = locAddr.GetEid();
628 2 : Hccl::Eid rmtEid = rmtAddr.GetEid();
629 :
630 : std::string dfxStr = Hccl::StringFormat(
631 : "chip id[%u] die id[%u] func_id[%u] jetty id[%s] "
632 : "local %s remote %s udp sport[%u]",
633 2 : devLogicId_, dieId_, funcId_, jettyIds.c_str(), locEid.Describe().c_str(), rmtEid.Describe().c_str(), udpSport);
634 2 : dfxMsg += dfxStr;
635 2 : HCCL_INFO("[CcuConnection::%s] %s", __func__, dfxStr.c_str());
636 2 : return HcclResult::HCCL_SUCCESS;
637 2 : }
638 :
639 15 : uint32_t CcuConnection::GetDieId() const { return dieId_; }
640 :
641 33 : uint32_t CcuConnection::GetChannelId() const { return channelInfo_.channelId; }
642 :
643 15 : int32_t CcuConnection::GetDevLogicId() const { return devLogicId_; }
644 :
645 0 : uint64_t CcuConnection::GetRmtCcuBufAddr() const { return rmtCcuBufAddr_; }
646 :
647 8 : uint32_t CcuConnection::GetRmtCcuBufTokenId() const { return rmtCcuBufTokenId_; }
648 :
649 8 : uint32_t CcuConnection::GetRmtCcuBufTokenValue() const { return rmtCcuBufTokenValue_; }
650 :
651 1 : HcclResult CcuConnection::Clean()
652 : {
653 1 : status_ = CcuConnStatus::INIT;
654 1 : innerStatus_ = InnerStatus::INIT;
655 1 : isJettyCreated_ = false;
656 1 : isJettyImported_ = false;
657 1 : CHK_RET(ReleaseConnRes());
658 1 : GenerateLocalPsn();
659 :
660 : // 销毁jetty要在ReleaseConnRes之后
661 1 : for (auto& ccuJetty : ccuJettys_) {
662 0 : ccuJetty->Clean();
663 : }
664 1 : return HcclResult::HCCL_SUCCESS;
665 : }
666 :
667 : } // namespace hcomm
|
