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_connection.h"
12 :
13 : #include <cstdlib>
14 : #include "hccp_ctx.h"
15 : #include "exception_util.h"
16 : #include "orion_adapter_rts.h"
17 : #include "internal_exception.h"
18 : #include "rdma_handle_manager.h"
19 :
20 : namespace Hccl {
21 :
22 269 : CcuConnection::CcuConnection(
23 : const IpAddress& locAddr, const IpAddress& rmtAddr, const CcuChannelInfo& channelInfo,
24 269 : const std::vector<CcuJetty*>& ccuJettys)
25 269 : : locAddr_(locAddr),
26 269 : rmtAddr_(rmtAddr),
27 269 : channelInfo_(channelInfo),
28 269 : ccuJettys_(ccuJettys)
29 269 : {}
30 :
31 1 : CcuTpConnection::CcuTpConnection(
32 : const IpAddress& locAddr, const IpAddress& rmtAddr, const CcuChannelInfo& channelInfo,
33 1 : const std::vector<CcuJetty*>& ccuJettys)
34 1 : : CcuConnection(locAddr, rmtAddr, channelInfo, ccuJettys)
35 : {
36 1 : tpProtocol = TpProtocol::TP;
37 1 : }
38 :
39 30 : CcuCtpConnection::CcuCtpConnection(
40 : const IpAddress& locAddr, const IpAddress& rmtAddr, const CcuChannelInfo& channelInfo,
41 30 : const std::vector<CcuJetty*>& ccuJettys)
42 30 : : CcuConnection(locAddr, rmtAddr, channelInfo, ccuJettys)
43 : {
44 30 : tpProtocol = TpProtocol::CTP;
45 30 : }
46 :
47 20 : HcclResult CcuConnection::Init()
48 : {
49 20 : TRY_CATCH_RETURN(devLogicId = HrtGetDevice(); uint32_t devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
50 :
51 : auto& rdmaHandleMgr = RdmaHandleManager::GetInstance();
52 : rdmaHandle = rdmaHandleMgr.GetByIp(devPhyId, locAddr_);
53 : dieId = rdmaHandleMgr.GetDieAndFuncId(rdmaHandle).first;);
54 :
55 20 : CHK_RET(GetLocalCcuRmaBufferInfo());
56 :
57 20 : jettyNum = channelInfo_.jettyInfos.size();
58 20 : CHK_PRT_RET(
59 : jettyNum == 0, HCCL_ERROR("[CcuConnection][%s] failed, jetty num[0] is unexpected.", __func__),
60 : HcclResult::HCCL_E_PARA);
61 :
62 20 : GenerateLocalPsn();
63 20 : status = CcuConnStatus::INIT;
64 20 : innerStatus = InnerStatus::INIT;
65 20 : return HcclResult::HCCL_SUCCESS;
66 : }
67 :
68 38 : CcuConnStatus CcuConnection::GetStatus()
69 : {
70 38 : if (status == CcuConnStatus::CONNECTED || status == CcuConnStatus::CONN_INVALID) {
71 3 : return status;
72 : }
73 :
74 35 : if (StatusMachine() != HcclResult::HCCL_SUCCESS) {
75 2 : status = CcuConnStatus::CONN_INVALID;
76 2 : innerStatus = InnerStatus::CONN_INVALID;
77 : }
78 :
79 35 : return status;
80 : }
81 : // 获取本端内存，为部分远端不可写的tokenId
82 20 : HcclResult CcuConnection::GetLocalCcuRmaBufferInfo()
83 : {
84 20 : uint64_t ccuBufSize = 0; // 暂未使用
85 20 : CHK_RET(CcuDeviceManager::GetCcuResourceSpaceBufInfo(devLogicId, dieId, ccuBufAddr, ccuBufSize));
86 :
87 20 : uint64_t tokenId = 0;
88 20 : uint64_t tokenValue = 0;
89 20 : CHK_RET(CcuDeviceManager::GetCcuResourceSpaceTokenInfo(devLogicId, dieId, tokenId, tokenValue));
90 20 : ccuBufTokenId = static_cast<uint32_t>(tokenId);
91 20 : ccuBufTokenValue = static_cast<uint32_t>(tokenValue);
92 20 : return HcclResult::HCCL_SUCCESS;
93 : }
94 :
95 35 : HcclResult CcuConnection::StatusMachine()
96 : {
97 161 : TRY_CATCH_RETURN(
98 : if (status == CcuConnStatus::INIT) {
99 : UpdateInitStatus();
100 : return HcclResult::HCCL_SUCCESS;
101 : }
102 :
103 : if (innerStatus == InnerStatus::JETTY_IMPORTING) {
104 : UpdateExchangeStatus();
105 : return HcclResult::HCCL_SUCCESS;
106 : });
107 :
108 0 : return HcclResult::HCCL_SUCCESS;
109 : }
110 :
111 29 : void CcuConnection::UpdateInitStatus()
112 : {
113 29 : switch (innerStatus) {
114 21 : case InnerStatus::INIT:
115 : case InnerStatus::JETTY_CREATING: {
116 21 : if (!CreateJetty()) {
117 10 : innerStatus = InnerStatus::JETTY_CREATING;
118 10 : break; // 状态不改变退出，下轮状态机进入继续执行
119 : }
120 :
121 10 : if (GetTpInfo()) { // 如果有缓存的tp信息，可以直接完成
122 1 : innerStatus = InnerStatus::EXCHANGEABLE;
123 1 : status = CcuConnStatus::EXCHANGEABLE;
124 1 : break;
125 : }
126 :
127 9 : innerStatus = InnerStatus::TP_INFO_GETTING; // 不退出继续调用下个异步接口
128 9 : break;
129 : }
130 8 : case InnerStatus::TP_INFO_GETTING: {
131 8 : if (!GetTpInfo()) {
132 0 : break; // 状态不改变退出，下轮状态机进入继续执行
133 : }
134 8 : innerStatus = InnerStatus::EXCHANGEABLE;
135 8 : status = CcuConnStatus::EXCHANGEABLE;
136 8 : break;
137 : }
138 0 : default:
139 0 : ThrowAbnormalStatus(std::string(__func__));
140 : }
141 28 : }
142 :
143 22 : bool CcuConnection::CreateJetty()
144 : {
145 22 : if (isJettyCreated) {
146 1 : return true;
147 : }
148 :
149 21 : isJettyCreated = true;
150 61 : for (size_t i = 0; i < jettyNum; i++) {
151 41 : auto ret = ccuJettys_[i]->CreateJetty();
152 41 : if (ret == HcclResult::HCCL_E_AGAIN) {
153 : // 不提供日志避免刷屏
154 10 : isJettyCreated = isJettyCreated && false;
155 10 : continue;
156 : }
157 :
158 31 : if (ret != HcclResult::HCCL_SUCCESS) {
159 3 : HCCL_ERROR("[CcuConnection][%s] failed, hccl result[%d]", __func__, ret);
160 2 : ThrowAbnormalStatus(std::string(__func__));
161 : }
162 : }
163 :
164 20 : return isJettyCreated;
165 : }
166 :
167 0 : inline uint32_t GetRandomNum()
168 : {
169 0 : uint32_t randNum = std::rand();
170 0 : return randNum;
171 : }
172 :
173 40 : void CcuConnection::GenerateLocalPsn() { jettyImportCfg.localPsn = GetRandomNum(); }
174 :
175 19 : bool CcuConnection::GetTpInfo()
176 : {
177 19 : if (tpProtocol == TpProtocol::INVALID) { // 不感知tp建链，当前默认不支持
178 3 : HCCL_ERROR(
179 : "[CcuConnection][%s] failed, tpProtocol[%s] is not expected.", __func__, tpProtocol.Describe().c_str());
180 2 : ThrowAbnormalStatus(std::string(__func__));
181 : }
182 :
183 18 : HcclResult ret = TpManager::GetInstance(devLogicId).GetTpInfo({locAddr_, rmtAddr_, tpProtocol}, tpInfo);
184 18 : if (ret == HcclResult::HCCL_E_AGAIN) {
185 : // 此处可能刷屏，非必要勿加日志
186 9 : return false;
187 : }
188 :
189 9 : if (ret != HcclResult::HCCL_SUCCESS) {
190 0 : HCCL_ERROR("[CcuConnection][%s] failed, hccl result[%d]", __func__, ret);
191 0 : ThrowAbnormalStatus(std::string(__func__));
192 : }
193 :
194 9 : jettyImportCfg.localTpHandle = tpInfo.tpHandle;
195 9 : return true;
196 : }
197 :
198 9 : void CcuConnection::Serialize(std::vector<char>& dtoData)
199 : {
200 9 : if (status != CcuConnStatus::EXCHANGEABLE) {
201 3 : HCCL_ERROR(
202 : "[CcuConnection][%s] failed, not init completed yet, "
203 : "status[%s].",
204 : __func__, status.Describe().c_str());
205 2 : ThrowAbnormalStatus(std::string(__func__));
206 : }
207 :
208 8 : BinaryStream dtoStream;
209 8 : dtoStream << ccuBufAddr;
210 8 : dtoStream << ccuBufTokenId;
211 8 : dtoStream << ccuBufTokenValue;
212 24 : HCCL_INFO("[CcuConnection][%s], ccuBufAddr[%llx]", __func__, ccuBufAddr);
213 :
214 8 : dtoStream << jettyNum;
215 24 : HCCL_INFO("[CcuConnection][%s], jettyNum[%u]", __func__, jettyNum);
216 24 : for (const auto& ccuJetty : ccuJettys_) {
217 16 : dtoStream << ccuJetty->GetCreateJettyParam().tokenValue;
218 16 : const auto& outParam = ccuJetty->GetJettyedOutParam();
219 16 : dtoStream << outParam.key;
220 16 : dtoStream << outParam.keySize;
221 : }
222 :
223 8 : if (tpProtocol != TpProtocol::INVALID) {
224 8 : dtoStream << jettyImportCfg.localTpHandle;
225 8 : dtoStream << jettyImportCfg.localPsn;
226 24 : HCCL_INFO(
227 : "[CcuConnection][%s] tpProtocol[%s], localTpHandle[0x%llx], localPsn[%u].", __func__,
228 : tpProtocol.Describe().c_str(), jettyImportCfg.localTpHandle, jettyImportCfg.localPsn);
229 : }
230 :
231 8 : dtoData.clear();
232 8 : dtoStream.Dump(dtoData);
233 8 : }
234 :
235 8 : void CcuConnection::Deserialize(const std::vector<char>& dtoData)
236 : {
237 8 : if (status != CcuConnStatus::EXCHANGEABLE) {
238 3 : HCCL_ERROR(
239 : "[CcuConnection][%s] failed, not init completed yet, "
240 : "status[%s].",
241 : __func__, status.Describe().c_str());
242 2 : ThrowAbnormalStatus(std::string(__func__));
243 : }
244 :
245 7 : vector<char> rmtDtoData = dtoData;
246 7 : BinaryStream dtoStream(rmtDtoData);
247 7 : dtoStream >> rmtCcuBufAddr;
248 7 : dtoStream >> rmtCcuBufTokenId;
249 7 : dtoStream >> rmtCcuBufTokenValue;
250 21 : HCCL_INFO("[CcuConnection][%s], rmtCcuBufAddr[%llx].", __func__, rmtCcuBufAddr);
251 :
252 7 : uint32_t remoteJettySize{0};
253 7 : dtoStream >> remoteJettySize;
254 :
255 7 : importJettyCtxs.clear();
256 7 : importJettyCtxs.resize(remoteJettySize);
257 21 : HCCL_INFO("[CcuConnection][%s], remoteJettySize[%u].", __func__, remoteJettySize);
258 :
259 21 : for (auto& importCtx : importJettyCtxs) {
260 14 : dtoStream >> importCtx.inParam.tokenValue;
261 14 : dtoStream >> importCtx.remoteQpKey; // 保存key数组
262 14 : importCtx.inParam.key = importCtx.remoteQpKey; // 保存指针用于接口调用
263 14 : dtoStream >> importCtx.inParam.keyLen;
264 : }
265 :
266 7 : if (tpProtocol != TpProtocol::INVALID) {
267 7 : dtoStream >> jettyImportCfg.remoteTpHandle;
268 7 : dtoStream >> jettyImportCfg.remotePsn;
269 :
270 21 : HCCL_INFO(
271 : "[CcuConnection][%s] tpEnable, remoteTpHandle[0x%llx], remotePsn[%u].", __func__,
272 : jettyImportCfg.remoteTpHandle, jettyImportCfg.remotePsn);
273 : }
274 7 : }
275 :
276 11 : void CcuConnection::ImportJetty()
277 : {
278 11 : if (isJettyImported) {
279 3 : HCCL_INFO("[CcuConnection][%s] taJettys has been imported already.", __func__);
280 1 : return;
281 : }
282 :
283 10 : if (innerStatus != InnerStatus::EXCHANGEABLE) {
284 4 : ThrowAbnormalStatus(std::string(__func__));
285 : }
286 :
287 8 : if (jettyNum != importJettyCtxs.size()) {
288 3 : HCCL_ERROR(
289 : "[CcuConnection][%s] failed to ImportJetty, "
290 : "jettyNum[%u] is not equal to importJettyCtxs.size[%u].",
291 : __func__, jettyNum, importJettyCtxs.size());
292 2 : ThrowAbnormalStatus(std::string(__func__));
293 : }
294 :
295 7 : ResetRequestCtxs();
296 19 : for (size_t i = 0; i < jettyNum; i++) {
297 13 : if (StartImportJettyRequest(i, reqHandles[i]) != HcclResult::HCCL_SUCCESS) {
298 2 : ThrowAbnormalStatus(std::string(__func__));
299 : }
300 : }
301 :
302 6 : innerStatus = InnerStatus::JETTY_IMPORTING;
303 : }
304 :
305 7 : void CcuConnection::ResetRequestCtxs()
306 : {
307 7 : reqHandles.clear();
308 7 : reqHandles.resize(jettyNum);
309 :
310 7 : reqDataBuffers.clear();
311 7 : reqDataBuffers.resize(jettyNum);
312 :
313 7 : remoteJettyHandlePtrs.clear();
314 7 : remoteJettyHandlePtrs.resize(jettyNum);
315 7 : }
316 :
317 12 : HcclResult CcuConnection::StartImportJettyRequest(uint32_t jettyIndex, RequestHandle& reqHandle)
318 : {
319 12 : if (tpProtocol == TpProtocol::INVALID) {
320 0 : ThrowAbnormalStatus(std::string(__func__));
321 : }
322 :
323 12 : auto& importCtx = importJettyCtxs[jettyIndex];
324 12 : auto& importCtxInParam = importCtx.inParam;
325 12 : importCtxInParam.jettyImportCfg = jettyImportCfg;
326 12 : importCtxInParam.jettyImportCfg.protocol = tpProtocol;
327 12 : TRY_CATCH_RETURN(reqHandle = RaUbTpImportJettyAsync(
328 : rdmaHandle, importCtxInParam, reqDataBuffers[jettyIndex], remoteJettyHandlePtrs[jettyIndex]););
329 :
330 12 : return HcclResult::HCCL_SUCCESS;
331 : }
332 :
333 6 : bool CcuConnection::CheckRequestResults()
334 : {
335 6 : if (reqHandles.size() == 0) {
336 0 : return true;
337 : }
338 :
339 : // 检查所有下发异步请求是否完成
340 6 : vector<size_t> completedReqs;
341 18 : for (size_t i = 0; i < reqHandles.size(); i++) {
342 12 : ReqHandleResult result = HrtRaGetAsyncReqResult(reqHandles[i]);
343 12 : if (result == ReqHandleResult::NOT_COMPLETED) {
344 0 : continue;
345 : }
346 :
347 12 : if (result != ReqHandleResult::COMPLETED) {
348 0 : THROW<InternalException>(StringFormat(
349 0 : "[CcuConnection][%s] failed, result[%s] is unexpected.", __func__, result.Describe().c_str()));
350 : }
351 :
352 : // 记录已完成的reqHandles
353 12 : completedReqs.push_back(i);
354 : }
355 :
356 : // 删除已完成的reqHandles，避免重复查询
357 18 : for (int i = completedReqs.size() - 1; i >= 0; --i) {
358 12 : reqHandles.erase(reqHandles.begin() + completedReqs[i]);
359 : }
360 :
361 : // 检查是否有剩余reqHandles
362 6 : return reqHandles.size() == 0;
363 6 : }
364 :
365 6 : void CcuConnection::UpdateExchangeStatus()
366 : {
367 : // 状态机保证为 InnerStatus::JETTY_IMPORTING
368 6 : if (!CheckRequestResults()) {
369 0 : return;
370 : }
371 :
372 18 : for (size_t i = 0; i < jettyNum; i++) {
373 12 : auto& outParam = importJettyCtxs[i].outParam;
374 12 : struct QpImportInfoT* infoPtr = reinterpret_cast<QpImportInfoT*>(reqDataBuffers[i].data());
375 12 : outParam.handle = reinterpret_cast<TargetJettyHandle>(remoteJettyHandlePtrs[i]);
376 12 : outParam.targetJettyVa = infoPtr->out.ub.tjettyHandle; // 该信息当前未使用
377 12 : outParam.tpn = infoPtr->out.ub.tpn;
378 : }
379 6 : isJettyImported = true;
380 :
381 6 : ConfigChannel();
382 5 : status = CcuConnStatus::CONNECTED;
383 5 : innerStatus = InnerStatus::CONNECTED;
384 : }
385 :
386 7 : void CcuConnection::ConfigChannel()
387 : {
388 7 : if (jettyNum != importJettyCtxs.size()) {
389 3 : HCCL_ERROR(
390 : "[CcuConnection][%s] failed, jettyNum[%u] is not equal to "
391 : "importJettyCtxs.size[%u].",
392 : __func__, jettyNum, importJettyCtxs.size());
393 2 : ThrowAbnormalStatus(std::string(__func__));
394 : }
395 :
396 6 : ChannelCfg cfg{};
397 6 : cfg.channelId = channelInfo_.channelId;
398 6 : cfg.remoteEid = rmtAddr_.GetReverseEid();
399 18 : HCCL_INFO("[CcuComponent::ConfigChannel] cfg.remoteEid=%s", cfg.remoteEid.Describe().c_str());
400 6 : cfg.tpn = importJettyCtxs[0].outParam.tpn; // tp handle复用所以tpn一致
401 6 : cfg.remoteCcuVa = rmtCcuBufAddr;
402 6 : cfg.memTokenId = rmtCcuBufTokenId;
403 6 : cfg.memTokenValue = rmtCcuBufTokenValue;
404 :
405 18 : for (size_t i = 0; i < jettyNum; i++) {
406 12 : const auto& ccuJetty = ccuJettys_[i];
407 12 : const auto& inParam = ccuJetty->GetCreateJettyParam();
408 12 : const auto& outParam = ccuJetty->GetJettyedOutParam();
409 12 : const auto& jettyInfo = channelInfo_.jettyInfos[i];
410 12 : cfg.jettyCfgs.emplace_back(JettyCfg{
411 12 : jettyInfo.jettyCtxId, outParam.dbVa, outParam.dbTokenId,
412 12 : inParam.tokenValue}); // 安全问题，禁止打印token相关信息
413 : }
414 :
415 6 : if (CcuDeviceManager::ConfigChannel(devLogicId, dieId, cfg) != HcclResult::HCCL_SUCCESS) {
416 3 : HCCL_ERROR(
417 : "[CcuConnection][%s] failed, devLogicId[%d], dieId[%u] channelId[%u].", __func__, devLogicId, dieId,
418 : cfg.channelId);
419 2 : ThrowAbnormalStatus(std::string(__func__));
420 : }
421 6 : }
422 :
423 269 : CcuConnection::~CcuConnection() { DECTOR_TRY_CATCH("CcuConnection", ReleaseConnRes()); }
424 :
425 258 : HcclResult CcuConnection::ReleaseConnRes()
426 : {
427 312 : for (auto& item : importJettyCtxs) {
428 54 : if (item.outParam.handle != 0) {
429 0 : HrtRaUbUnimportJetty(rdmaHandle, item.outParam.handle);
430 0 : item.outParam.handle = 0;
431 : }
432 : }
433 :
434 258 : if (tpInfo.tpHandle != 0) { // tp handle 复用，只释放一次
435 1 : (void)TpManager::GetInstance(devLogicId).ReleaseTpInfo({locAddr_, rmtAddr_, tpProtocol}, tpInfo);
436 1 : tpInfo.tpHandle = 0;
437 : }
438 :
439 : // CcuJetty 生命周期跟随通信域CcuJettyMgr
440 : // 不需要connection主动销毁
441 258 : return HcclResult::HCCL_SUCCESS;
442 : }
443 :
444 10 : void CcuConnection::ThrowAbnormalStatus(const string& funcName)
445 : {
446 10 : auto errMsg = StringFormat("[CcuConnection][%s] failed, [%s].", funcName.c_str(), Describe().c_str());
447 10 : status = CcuConnStatus::CONN_INVALID;
448 10 : innerStatus = InnerStatus::CONN_INVALID;
449 10 : THROW<InternalException>(errMsg);
450 10 : }
451 :
452 10 : std::string CcuConnection::Describe()
453 : {
454 : return StringFormat(
455 : "[CcuConnection[locAddr=%s, rmtAddr=%s, protocol=%s, "
456 : "status=%s, innerStatus=%s, [dieId=%u, channelId=%u, jettyNum=%u]]]",
457 40 : locAddr_.Describe().c_str(), rmtAddr_.Describe().c_str(), tpProtocol.Describe().c_str(),
458 50 : status.Describe().c_str(), innerStatus.Describe().c_str(), dieId, channelInfo_.channelId, jettyNum);
459 : }
460 :
461 9 : uint32_t CcuConnection::GetDieId() const { return dieId; }
462 :
463 648 : uint32_t CcuConnection::GetChannelId() const { return channelInfo_.channelId; }
464 :
465 9 : int32_t CcuConnection::GetDevLogicId() const { return devLogicId; }
466 :
467 19 : std::vector<ConnJettyInfo> CcuConnection::GetDeleteJettyInfo()
468 : {
469 19 : std::vector<ConnJettyInfo> connDeleteJettyInfos;
470 19 : ConnJettyInfo jettyInfo;
471 43 : for (auto& ccuJetty : ccuJettys_) {
472 24 : if (ccuJetty != nullptr) {
473 24 : ccuJetty->GetJettyInfo(jettyInfo);
474 24 : jettyInfo.rdmaHandle = rdmaHandle;
475 24 : connDeleteJettyInfos.push_back(jettyInfo);
476 : }
477 : }
478 19 : return connDeleteJettyInfos;
479 0 : }
480 :
481 19 : std::vector<ConnJettyInfo> CcuConnection::GetUnimportJettyInfo()
482 : {
483 19 : std::vector<ConnJettyInfo> connUnimportJettyInfos;
484 19 : ConnJettyInfo jettyInfo;
485 43 : for (auto& item : importJettyCtxs) {
486 24 : if (item.outParam.handle != 0) {
487 14 : jettyInfo.remoteJetty = item.outParam.handle;
488 14 : jettyInfo.rdmaHandle = rdmaHandle;
489 14 : item.outParam.handle = 0;
490 14 : connUnimportJettyInfos.push_back(jettyInfo);
491 : }
492 : }
493 19 : return connUnimportJettyInfos;
494 0 : }
495 :
496 20 : void CcuConnection::Clean()
497 : {
498 20 : status = CcuConnStatus::INIT;
499 20 : innerStatus = InnerStatus::INIT;
500 20 : isJettyCreated = false;
501 20 : isJettyImported = false;
502 20 : ReleaseConnRes();
503 20 : GenerateLocalPsn();
504 :
505 : // 销毁jetty要在ReleaseConnRes之后
506 46 : for (auto& ccuJetty : ccuJettys_) {
507 26 : ccuJetty->Clean();
508 : }
509 20 : }
510 :
511 : } // namespace Hccl
|
