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