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_transport.h"
12 : #include "coll_operator_check.h"
13 : #include "exception_util.h"
14 :
15 : namespace Hccl {
16 :
17 : constexpr uint32_t FINISH_MSG_SIZE = 128;
18 : constexpr char_t FINISH_MSG[FINISH_MSG_SIZE] = "Transport exchange data ready!";
19 :
20 11 : HcclResult CcuCreateTransport(
21 : Socket* socket, const CcuTransport::CcuConnectionInfo& ccuConnectionInfo,
22 : const CcuTransport::CclBufferInfo& cclBufferInfo, std::unique_ptr<CcuTransport>& ccuTransport)
23 : {
24 11 : CHK_PTR_NULL(socket);
25 33 : HCCL_INFO(
26 : "[%s]ccuConnectionInfo type[%d], locAddr[%s], rmtAddr[%s], channelInfo[channelId %u:dieId %u], "
27 : "cclBufferInfo addr[%llu], size[%u]",
28 : __func__, ccuConnectionInfo.type, ccuConnectionInfo.locAddr.GetIpStr().c_str(),
29 : ccuConnectionInfo.rmtAddr.GetIpStr().c_str(), ccuConnectionInfo.channelInfo.channelId,
30 : ccuConnectionInfo.channelInfo.dieId, cclBufferInfo.addr, cclBufferInfo.size);
31 11 : TRY_CATCH_RETURN(
32 : std::unique_ptr<CcuConnection> ccuConnection;
33 : if (ccuConnectionInfo.type == CcuTransport::CcuConnectionType::UB_CTP) {
34 : ccuConnection = std::make_unique<CcuCtpConnection>(
35 : ccuConnectionInfo.locAddr, ccuConnectionInfo.rmtAddr, ccuConnectionInfo.channelInfo,
36 : ccuConnectionInfo.ccuJettys);
37 : } else {
38 : ccuConnection = std::make_unique<CcuTpConnection>(
39 : ccuConnectionInfo.locAddr, ccuConnectionInfo.rmtAddr, ccuConnectionInfo.channelInfo,
40 : ccuConnectionInfo.ccuJettys);
41 : }
42 :
43 : auto ret
44 : = ccuConnection->Init();
45 : if (ret != HcclResult::HCCL_SUCCESS) {
46 : ccuConnection = nullptr;
47 : return ret;
48 : }
49 :
50 : ccuTransport
51 : = std::make_unique<CcuTransport>(socket, std::move(ccuConnection), cclBufferInfo);
52 : ret = ccuTransport->Init(); if (ret != HcclResult::HCCL_SUCCESS) {
53 : ccuTransport = nullptr;
54 : return ret;
55 : });
56 :
57 11 : return HcclResult::HCCL_SUCCESS;
58 : }
59 :
60 262 : CcuTransport::CcuTransport(
61 262 : Socket* socket, std::unique_ptr<CcuConnection>&& connection, const CclBufferInfo& locCclBufInfo)
62 262 : : socket(socket),
63 262 : ccuConnection(std::move(connection)),
64 1048 : locCclBufInfo(locCclBufInfo)
65 262 : {}
66 :
67 9 : HcclResult CcuTransport::Init()
68 : {
69 9 : dieId = ccuConnection->GetDieId();
70 9 : devLogicId = ccuConnection->GetDevLogicId();
71 9 : auto ret = AppendCkes(INIT_CKE_NUM);
72 9 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
73 3 : HCCL_WARNING("[AppendCkes] UNAVAIL.");
74 1 : return ret;
75 : }
76 :
77 8 : if (ret != HCCL_SUCCESS) {
78 3 : HCCL_ERROR("errNo[0x%016llx]:AppendCkes failed.", ret);
79 1 : return ret;
80 : }
81 7 : ret = AppendXns(INIT_XN_NUM);
82 7 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
83 3 : HCCL_WARNING("[AppendXns] UNAVAIL.");
84 1 : return ret;
85 : }
86 :
87 6 : if (ret != HcclResult::HCCL_SUCCESS) {
88 0 : HCCL_ERROR("errNo[0x%016llx]:AppendXns failed.", ret);
89 0 : return ret;
90 : }
91 6 : transStatus = TransStatus::INIT;
92 6 : return HCCL_SUCCESS;
93 : }
94 :
95 184 : HcclResult CcuTransport::AppendRes(uint32_t ckesNum, uint32_t xnsNum)
96 : {
97 : try {
98 184 : std::unique_lock<std::shared_timed_mutex> lock(transMutex);
99 184 : auto ret = AppendCkes(ckesNum);
100 184 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
101 3 : HCCL_WARNING("[AppendCkes] UNAVAIL.");
102 1 : return ret;
103 : }
104 :
105 183 : if (ret != HCCL_SUCCESS) {
106 462 : HCCL_ERROR("errNo[0x%016llx]:AppendCkes failed.", ret);
107 154 : return ret;
108 : }
109 29 : ret = AppendXns(xnsNum);
110 29 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
111 3 : HCCL_WARNING("[AppendXns] UNAVAIL.");
112 1 : return ret;
113 : }
114 :
115 28 : if (ret != HcclResult::HCCL_SUCCESS) {
116 0 : HCCL_ERROR("errNo[0x%016llx]:AppendXns failed.", ret);
117 0 : return ret;
118 : }
119 28 : transStatus = CcuTransport::TransStatus::SEND_TRANS_RES;
120 184 : } catch (HcclException& e) {
121 0 : HCCL_ERROR(e.what());
122 0 : return HCCL_E_INTERNAL;
123 0 : } catch (exception& e) {
124 0 : HCCL_ERROR(e.what());
125 0 : return HCCL_E_INTERNAL;
126 0 : } catch (...) {
127 0 : HCCL_ERROR("Unknown error occured during unimport jetty or destroy jetty!");
128 0 : return HCCL_E_INTERNAL;
129 0 : }
130 28 : return HCCL_SUCCESS;
131 : }
132 :
133 193 : HcclResult CcuTransport::AppendCkes(uint32_t ckesNum)
134 : {
135 193 : vector<ResInfo> resInfo;
136 193 : auto ret = CcuDeviceManager::AllocCke(devLogicId, dieId, ckesNum, resInfo);
137 193 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
138 6 : HCCL_WARNING("[AppendCkes] UNAVAIL.");
139 2 : return ret;
140 : }
141 :
142 191 : if (ret != HcclResult::HCCL_SUCCESS) {
143 465 : HCCL_ERROR("errNo[0x%016llx]:AppendCkes failed.", ret);
144 155 : return ret;
145 : }
146 :
147 72 : for (uint32_t i = 0; i < resInfo.size(); i++) {
148 36 : uint32_t ckeNum = resInfo[i].num;
149 36 : uint32_t ckesSartId = resInfo[i].startId;
150 239 : for (uint32_t j = 0; j < ckeNum; j++) {
151 203 : locRes.ckes.push_back(ckesSartId + j);
152 : }
153 : }
154 36 : ckesRes.push_back(resInfo);
155 36 : return HCCL_SUCCESS;
156 193 : }
157 :
158 36 : HcclResult CcuTransport::AppendXns(uint32_t xnsNum)
159 : {
160 36 : vector<ResInfo> resInfo;
161 36 : auto ret = CcuDeviceManager::AllocXn(devLogicId, dieId, xnsNum, resInfo);
162 36 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
163 6 : HCCL_WARNING("[AppendXns] UNAVAIL.");
164 2 : return ret;
165 : }
166 :
167 34 : if (ret != HcclResult::HCCL_SUCCESS) {
168 0 : HCCL_ERROR("errNo[0x%016llx]:AppendXns failed.", ret);
169 0 : return ret;
170 : }
171 68 : for (uint32_t i = 0; i < resInfo.size(); i++) {
172 34 : uint32_t xnNum = resInfo[i].num;
173 34 : uint32_t xnsSartId = resInfo[i].startId;
174 214 : for (uint32_t j = 0; j < xnNum; j++) {
175 180 : locRes.xns.push_back(xnsSartId + j);
176 : }
177 : }
178 34 : xnsRes.push_back(resInfo);
179 34 : return HCCL_SUCCESS;
180 36 : }
181 :
182 187 : void CcuTransport::SetCntCke(const vector<uint32_t>& cntCke)
183 : {
184 561 : HCCL_INFO("[%s]cntCke size[%llu]", __func__, cntCke.size());
185 187 : locRes.cntCkes = cntCke;
186 187 : }
187 :
188 19 : CcuTransport::TransStatus CcuTransport::StateMachine()
189 : {
190 19 : if (transStatus == TransStatus::READY) {
191 1 : return transStatus;
192 : }
193 18 : SocketStatus socketStatus = socket->GetAsyncStatus();
194 18 : if (socketStatus == SocketStatus::INIT) {
195 1 : THROW<InternalException>("[CcuTransport][GetStatus]socket timeout or no link, please check");
196 : }
197 :
198 17 : if (socketStatus == SocketStatus::TIMEOUT) {
199 1 : return TransStatus::SOCKET_TIMEOUT;
200 : }
201 :
202 16 : if (socketStatus != SocketStatus::OK) {
203 0 : return transStatus;
204 : }
205 :
206 16 : switch (transStatus) {
207 9 : case CcuTransport::TransStatus::INIT: {
208 9 : auto connStatus = ccuConnection->GetStatus();
209 9 : if (connStatus == CcuConnStatus::CONN_INVALID) {
210 1 : THROW<InternalException>(
211 : "[CcuTransport][GetStatus] connection status[%s] failed."
212 : " please check",
213 3 : connStatus.Describe().c_str());
214 : }
215 :
216 8 : if (connStatus == CcuConnStatus::EXCHANGEABLE || connStatus == CcuConnStatus::CONNECTED) {
217 2 : transStatus = CcuTransport::TransStatus::SEND_ALL_INFO;
218 2 : SendConnAndTransInfo();
219 : }
220 :
221 8 : break;
222 : }
223 2 : case CcuTransport::TransStatus::SEND_ALL_INFO:
224 2 : transStatus = CcuTransport::TransStatus::RECV_ALL_INFO;
225 2 : RecvConnAndTransInfo();
226 2 : break;
227 2 : case CcuTransport::TransStatus::RECV_ALL_INFO:
228 2 : RecvDataProcess();
229 1 : ccuConnection->ImportJetty();
230 1 : transStatus = CcuTransport::TransStatus::SEND_FIN;
231 1 : break;
232 1 : case CcuTransport::TransStatus::SEND_FIN: {
233 1 : auto connStatus = ccuConnection->GetStatus();
234 1 : if (connStatus == CcuConnStatus::CONN_INVALID) {
235 0 : THROW<InternalException>(
236 : "[CcuTransport][GetStatus] connection status[%s] failed."
237 : " please check",
238 0 : connStatus.Describe().c_str());
239 : }
240 :
241 1 : if (connStatus == CcuConnStatus::CONNECTED) {
242 1 : SendFinish();
243 1 : transStatus = CcuTransport::TransStatus::RECVING_FIN;
244 : }
245 1 : break;
246 : }
247 1 : case CcuTransport::TransStatus::RECVING_FIN:
248 1 : RecvFinish();
249 1 : transStatus = CcuTransport::TransStatus::RECV_FIN;
250 1 : break;
251 1 : case CcuTransport::TransStatus::RECV_FIN:
252 1 : CheckFinish();
253 1 : transStatus = CcuTransport::TransStatus::READY;
254 1 : break;
255 0 : case CcuTransport::TransStatus::SEND_TRANS_RES:
256 0 : SendTransInfo();
257 0 : transStatus = CcuTransport::TransStatus::RECVING_TRANS_RES;
258 0 : break;
259 0 : case CcuTransport::TransStatus::RECVING_TRANS_RES:
260 0 : RecvTransInfo();
261 0 : transStatus = CcuTransport::TransStatus::RECV_TRANS_RES;
262 0 : break;
263 0 : case CcuTransport::TransStatus::RECV_TRANS_RES:
264 0 : RecvTransInfoProcess();
265 0 : transStatus = CcuTransport::TransStatus::SEND_FIN;
266 0 : break;
267 0 : default:
268 0 : THROW<InternalException>("[CcuTransport][GetStatus]failed status");
269 : break;
270 : }
271 14 : return transStatus;
272 : }
273 :
274 19 : CcuTransport::TransStatus CcuTransport::GetStatus()
275 : {
276 19 : CcuTransport::TransStatus status = CcuTransport::TransStatus::CONNECT_FAILED;
277 19 : auto lockAndStatuMachine = [&]() {
278 19 : std::unique_lock<std::shared_timed_mutex> lock(transMutex);
279 19 : status = StateMachine();
280 19 : };
281 28 : TRY_CATCH_PROCESS_THROW(
282 : InternalException, lockAndStatuMachine(), "CcuTransport GetStatus() Error when creating transport connection",
283 : { transStatus = CcuTransport::TransStatus::CONNECT_FAILED; });
284 16 : return status;
285 : }
286 :
287 2 : void CcuTransport::SendConnAndTransInfo()
288 : {
289 2 : BinaryStream binaryStream;
290 2 : HandshakeMsgPack(binaryStream);
291 2 : ConnInfoPack(binaryStream);
292 2 : TransResPack(binaryStream);
293 2 : CclBufferInfoPack(binaryStream);
294 2 : binaryStream.Dump(sendData);
295 2 : socket->SendAsync(sendData.data(), sendData.size());
296 2 : exchangeDataSize = sendData.size();
297 2 : }
298 :
299 2 : void CcuTransport::RecvConnAndTransInfo()
300 : {
301 2 : recvData.resize(exchangeDataSize);
302 2 : socket->RecvAsync(reinterpret_cast<u8*>(recvData.data()), recvData.size());
303 2 : }
304 :
305 2 : void CcuTransport::RecvDataProcess()
306 : {
307 2 : BinaryStream binaryStream(recvData);
308 2 : HandshakeMsgUnpack(binaryStream);
309 1 : ConnInfoUnpackProc(binaryStream);
310 1 : TransResUnpackProc(binaryStream);
311 1 : CclBufferInfoUnpack(binaryStream);
312 2 : }
313 :
314 0 : void CcuTransport::SendTransInfo()
315 : {
316 0 : BinaryStream binaryStream;
317 0 : TransResPack(binaryStream);
318 0 : binaryStream.Dump(sendTrans);
319 0 : socket->SendAsync(sendTrans.data(), sendTrans.size());
320 0 : exchangeDataSize = sendTrans.size();
321 0 : }
322 :
323 0 : void CcuTransport::RecvTransInfo()
324 : {
325 0 : recvTrans.resize(exchangeDataSize);
326 0 : socket->RecvAsync(reinterpret_cast<u8*>(recvTrans.data()), recvTrans.size());
327 0 : }
328 :
329 0 : void CcuTransport::RecvTransInfoProcess()
330 : {
331 0 : BinaryStream binaryStream(recvTrans);
332 0 : TransResUnpackProc(binaryStream);
333 0 : }
334 :
335 2 : void CcuTransport::HandshakeMsgPack(BinaryStream& binaryStream)
336 : {
337 2 : binaryStream << static_cast<u32>(attr.opAcceState);
338 2 : binaryStream << attr.handshakeMsg;
339 6 : HCCL_INFO(
340 : "[CcuTransport][%s] start pack handshakeMsg, attr.handshakeMsg.size[%zu]", __func__, attr.handshakeMsg.size());
341 2 : }
342 :
343 2 : void CcuTransport::ConnInfoPack(BinaryStream& binaryStream) const
344 : {
345 2 : std::vector<char> dtoData{};
346 2 : ccuConnection->Serialize(dtoData);
347 2 : binaryStream << dtoData;
348 6 : HCCL_INFO("[CcuTransport][%s] start pack connInfo, dtoData.size[%u]", __func__, dtoData.size());
349 2 : }
350 :
351 2 : void CcuTransport::TransResPack(BinaryStream& binaryStream)
352 : {
353 2 : uint32_t locCkesSize = locRes.ckes.size();
354 2 : binaryStream << locCkesSize;
355 34 : for (uint32_t i = 0; i < locRes.ckes.size(); i++) {
356 32 : binaryStream << locRes.ckes[i];
357 : }
358 :
359 2 : uint32_t locCntCkesSize = locRes.cntCkes.size();
360 2 : binaryStream << locCntCkesSize;
361 34 : for (uint32_t i = 0; i < locRes.cntCkes.size(); i++) {
362 32 : binaryStream << locRes.cntCkes[i];
363 : }
364 :
365 2 : uint32_t locXnsSize = locRes.xns.size();
366 2 : binaryStream << locXnsSize;
367 34 : for (uint32_t i = 0; i < locRes.xns.size(); i++) {
368 32 : binaryStream << locRes.xns[i];
369 : }
370 6 : HCCL_INFO("Send ckesSize[%u], cntCkesSize[%u], xnsSize[%u]", locCkesSize, locCntCkesSize, locXnsSize);
371 2 : }
372 :
373 2 : void CcuTransport::CclBufferInfoPack(BinaryStream& binaryStream) const { locCclBufInfo.Pack(binaryStream); }
374 :
375 2 : void CcuTransport::HandshakeMsgUnpack(BinaryStream& binaryStream)
376 : {
377 2 : u32 rmtAccelerator{0};
378 2 : binaryStream >> rmtAccelerator;
379 6 : HCCL_INFO("[CcuTransport::HandshakeMsgUnpack], rmtAccelerator[%u]", rmtAccelerator);
380 2 : rmtOpAcceState = static_cast<AcceleratorState::Value>(rmtAccelerator);
381 :
382 2 : if (rmtOpAcceState != attr.opAcceState) {
383 0 : THROW<InvalidParamsException>(StringFormat(
384 : "[CcuTransport::HandshakeMsgUnpack] Accelerator information check fail. "
385 : "locOpAccelerator[%s], rmtOpAccelerator[%s]",
386 0 : attr.opAcceState.Describe().c_str(), rmtOpAcceState.Describe().c_str()));
387 : }
388 :
389 2 : rmtHandshakeMsg.clear();
390 2 : binaryStream >> rmtHandshakeMsg;
391 :
392 2 : if (attr.handshakeMsg.size() != rmtHandshakeMsg.size()) {
393 4 : MACRO_THROW(
394 : InvalidParamsException,
395 : StringFormat(
396 : "handshakeMsg size=%u is not equal to rmt=%u", attr.handshakeMsg.size(), rmtHandshakeMsg.size()));
397 : }
398 :
399 1 : auto localCollOperator = CollOperator::GetPackedData(attr.handshakeMsg);
400 1 : auto remoteCollOperator = CollOperator::GetPackedData(rmtHandshakeMsg);
401 1 : CheckCollOperator(localCollOperator, remoteCollOperator); // 两端算子参数一致性校验
402 :
403 3 : HCCL_INFO("[CcuTransport][%s] start unpack handshakeMsg", __func__);
404 1 : }
405 :
406 1 : void CcuTransport::ConnInfoUnpackProc(BinaryStream& binaryStream) const
407 : {
408 1 : std::vector<char> dtoData{};
409 1 : binaryStream >> dtoData;
410 1 : ccuConnection->Deserialize(dtoData);
411 3 : HCCL_INFO("[CcuTransport][%s] start unpack connInfo, dtoData.size[%zu]", __func__, dtoData.size());
412 1 : }
413 :
414 1 : void CcuTransport::TransResUnpackProc(BinaryStream& binaryStream)
415 : {
416 : uint32_t resSzie;
417 1 : binaryStream >> resSzie;
418 1 : rmtRes.ckes.clear();
419 17 : for (uint32_t i = 0; i < resSzie; i++) {
420 : uint32_t cke;
421 16 : binaryStream >> cke;
422 16 : rmtRes.ckes.push_back(cke);
423 : }
424 3 : HCCL_INFO("Recv ckesSize[%u]", resSzie);
425 :
426 1 : binaryStream >> resSzie;
427 1 : rmtRes.cntCkes.clear();
428 17 : for (uint32_t i = 0; i < resSzie; i++) {
429 : uint32_t cntCke;
430 16 : binaryStream >> cntCke;
431 16 : rmtRes.cntCkes.push_back(cntCke);
432 : }
433 :
434 1 : binaryStream >> resSzie;
435 1 : rmtRes.xns.clear();
436 17 : for (uint32_t i = 0; i < resSzie; i++) {
437 : uint32_t xn;
438 16 : binaryStream >> xn;
439 16 : rmtRes.xns.push_back(xn);
440 : }
441 3 : HCCL_INFO("Recv xnsSize[%u]", resSzie);
442 1 : }
443 :
444 1 : void CcuTransport::CclBufferInfoUnpack(BinaryStream& binaryStream) { rmtCclBufInfo.Unpack(binaryStream); }
445 :
446 1 : void CcuTransport::SendFinish()
447 : {
448 1 : sendFinishMsg = std::vector<char>(FINISH_MSG, FINISH_MSG + FINISH_MSG_SIZE);
449 1 : socket->SendAsync(sendFinishMsg.data(), FINISH_MSG_SIZE);
450 1 : }
451 :
452 1 : void CcuTransport::RecvFinish()
453 : {
454 1 : recvFinishMsg.resize(FINISH_MSG_SIZE);
455 1 : socket->RecvAsync(reinterpret_cast<u8*>(recvFinishMsg.data()), FINISH_MSG_SIZE);
456 1 : }
457 :
458 1 : void CcuTransport::CheckFinish()
459 : {
460 2 : std::string sendFinishMsgStr(sendFinishMsg.begin(), sendFinishMsg.end());
461 1 : std::string recvFinishMsgStr(recvFinishMsg.begin(), recvFinishMsg.end());
462 1 : if (sendFinishMsgStr != recvFinishMsgStr) {
463 0 : THROW<InternalException>(
464 : "[CcuTransport][RecvFinish]msgRecv[%s] and msgSend[%s] are not equal", recvFinishMsgStr.c_str(),
465 : sendFinishMsgStr.c_str());
466 : }
467 1 : }
468 :
469 229 : void CcuTransport::ReleaseTransRes()
470 : {
471 237 : for (uint32_t i = 0; i < ckesRes.size(); i++) {
472 8 : auto ret = CcuDeviceManager::ReleaseCke(devLogicId, dieId, ckesRes[i]);
473 8 : if (ret != HcclResult::HCCL_SUCCESS) {
474 0 : THROW<InternalException>("errNo[0x%016llx]:Release ckesRes failed.", ret);
475 : }
476 : }
477 :
478 235 : for (uint32_t i = 0; i < xnsRes.size(); i++) {
479 6 : auto ret = CcuDeviceManager::ReleaseXn(devLogicId, dieId, xnsRes[i]);
480 6 : if (ret != HcclResult::HCCL_SUCCESS) {
481 0 : THROW<InternalException>("errNo[0x%016llx]:Release xnsRes failed.", ret);
482 : }
483 : }
484 229 : }
485 :
486 410 : uint32_t CcuTransport::GetDieId() const { return dieId; }
487 :
488 648 : uint32_t CcuTransport::GetChannelId() const { return ccuConnection->GetChannelId(); }
489 :
490 5 : uint32_t CcuTransport::GetLocCkeByIndex(uint32_t index) const
491 : {
492 5 : std::shared_lock<std::shared_timed_mutex> lock(transMutex);
493 5 : if (index >= locRes.ckes.size()) {
494 0 : THROW<InternalException>(
495 : "[GetLocCkeByIndex]:index[%u] is bigger than ckes size[%u]", index, locRes.ckes.size());
496 : }
497 10 : return locRes.ckes[index];
498 5 : }
499 :
500 36 : uint32_t CcuTransport::GetLocCntCkeByIndex(uint32_t index) const
501 : {
502 36 : std::shared_lock<std::shared_timed_mutex> lock(transMutex);
503 36 : if (index >= locRes.cntCkes.size()) {
504 0 : THROW<InternalException>(
505 : "[GetLocCntCkeByIndex]:index[%u] is bigger than cntCkes size[%u]", index, locRes.cntCkes.size());
506 : }
507 72 : return locRes.cntCkes[index];
508 36 : }
509 :
510 10 : uint32_t CcuTransport::GetLocXnByIndex(uint32_t index) const
511 : {
512 10 : std::shared_lock<std::shared_timed_mutex> lock(transMutex);
513 10 : if (index >= locRes.xns.size()) {
514 0 : THROW<InternalException>(
515 0 : StringFormat("[GetLocXnByIndex]:index[%u] is bigger than xns size[%u]", index, locRes.xns.size()));
516 : }
517 20 : return locRes.xns[index];
518 10 : }
519 :
520 0 : uint32_t CcuTransport::GetRmtCkeByIndex(uint32_t index) const
521 : {
522 0 : std::shared_lock<std::shared_timed_mutex> lock(transMutex);
523 0 : if (index >= rmtRes.ckes.size()) {
524 0 : THROW<InternalException>(
525 0 : StringFormat("[GetRmtCkeByIndex]:index[%u] is bigger than ckes size[%u]", index, rmtRes.ckes.size()));
526 : }
527 0 : return rmtRes.ckes[index];
528 0 : }
529 :
530 66 : uint32_t CcuTransport::GetRmtCntCkeByIndex(uint32_t index) const
531 : {
532 66 : std::shared_lock<std::shared_timed_mutex> lock(transMutex);
533 66 : if (index >= rmtRes.cntCkes.size()) {
534 0 : THROW<InternalException>(StringFormat(
535 : "[GetRmtCntCkeByIndex]:index[%u] is bigger than cntCkes size[%u]", index, rmtRes.cntCkes.size()));
536 : }
537 132 : return rmtRes.cntCkes[index];
538 66 : }
539 :
540 40 : uint32_t CcuTransport::GetRmtXnByIndex(uint32_t index) const
541 : {
542 40 : std::shared_lock<std::shared_timed_mutex> lock(transMutex);
543 40 : if (index >= rmtRes.xns.size()) {
544 0 : THROW<InternalException>(
545 0 : StringFormat("[GetRmtXnByIndex]:index[%u] is bigger than xns size[%u]", index, rmtRes.xns.size()));
546 : }
547 80 : return rmtRes.xns[index];
548 40 : }
549 :
550 0 : HcclResult CcuTransport::GetLocBuffer(CclBufferInfo& bufferInfo, const uint32_t& bufNum) const
551 : {
552 : (void)bufNum;
553 0 : bufferInfo = locCclBufInfo;
554 0 : return HCCL_SUCCESS;
555 : }
556 :
557 0 : HcclResult CcuTransport::GetRmtBuffer(CclBufferInfo& bufferInfo, const uint32_t& bufNum) const
558 : {
559 : (void)bufNum;
560 0 : bufferInfo = rmtCclBufInfo;
561 0 : return HCCL_SUCCESS;
562 : }
563 :
564 262 : CcuTransport::~CcuTransport() { DECTOR_TRY_CATCH("CcuTransport", ReleaseTransRes()); }
565 :
566 15 : std::string CcuTransport::Describe() const
567 : {
568 15 : std::string description = "";
569 :
570 15 : description = StringFormat("DieId: %u, ", dieId);
571 15 : description += transStatus.Describe();
572 30 : description += StringFormat(
573 15 : ", LocRes: {%u Ckes, %u CntCkes, %u Xns}, ", locRes.ckes.size(), locRes.cntCkes.size(), locRes.xns.size());
574 30 : description += StringFormat(
575 15 : "RmtRes: {%u Ckes, %u CntCkes, %u Xns}, ", rmtRes.ckes.size(), rmtRes.cntCkes.size(), rmtRes.xns.size());
576 15 : description += StringFormat("CkesRes size: %u, ", ckesRes.size());
577 15 : description += StringFormat("XnsRes size: %u, ", xnsRes.size());
578 15 : description += StringFormat("%s", socket->Describe().c_str());
579 15 : return description;
580 0 : }
581 :
582 19 : std::vector<ConnJettyInfo> CcuTransport::GetDeleteJettyInfo() { return ccuConnection->GetDeleteJettyInfo(); }
583 19 : std::vector<ConnJettyInfo> CcuTransport::GetUnimportJettyInfo() { return ccuConnection->GetUnimportJettyInfo(); }
584 :
585 19 : HcclResult CcuTransport::Clean()
586 : {
587 19 : transStatus = TransStatus::INIT;
588 19 : sendData.clear();
589 19 : TRY_CATCH_RETURN(ccuConnection->Clean());
590 19 : return HCCL_SUCCESS;
591 : }
592 :
593 : } // namespace Hccl
|
