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