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