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 :
13 : #include "exception_handler.h"
14 :
15 : #include "../../../../../legacy/ascend950/unified_platform/resource/mem/user_remote_mem_getter.h"
16 :
17 : #include "env_config/env_config_v2.h"
18 :
19 : namespace hcomm {
20 :
21 : constexpr uint32_t FINISH_MSG_SIZE = 128;
22 : constexpr char FINISH_MSG[FINISH_MSG_SIZE] = "Transport exchange data ready!";
23 :
24 15 : HcclResult BuildCcuConnection(
25 : const CcuTransport::CcuConnectionInfo& ccuConnectionInfo, std::unique_ptr<CcuConnection>& ccuConnection)
26 : {
27 15 : if (ccuConnectionInfo.type == CcuTransport::CcuConnectionType::UB_CTP) {
28 30 : ccuConnection.reset(new (std::nothrow) CcuCtpConnection(
29 15 : ccuConnectionInfo.locAddr, ccuConnectionInfo.rmtAddr, ccuConnectionInfo.channelInfo,
30 30 : ccuConnectionInfo.ccuJettys, ccuConnectionInfo.qos));
31 : } else {
32 0 : ccuConnection.reset(new (std::nothrow) CcuRtpConnection(
33 0 : ccuConnectionInfo.locAddr, ccuConnectionInfo.rmtAddr, ccuConnectionInfo.channelInfo,
34 0 : ccuConnectionInfo.ccuJettys, ccuConnectionInfo.qos));
35 : }
36 15 : CHK_PTR_NULL(ccuConnection);
37 15 : CHK_RET(ccuConnection->Init());
38 15 : return HCCL_SUCCESS;
39 : }
40 :
41 0 : HcclResult CcuCreateTransport(
42 : Hccl::Socket* socket, const CcuTransport::CcuConnectionInfo& ccuConnectionInfo,
43 : const CcuTransport::CclBufferInfo& cclBufferInfo, std::unique_ptr<CcuTransport>& ccuTransport)
44 : {
45 0 : CHK_PTR_NULL(socket);
46 0 : std::unique_ptr<CcuConnection> ccuConnection{nullptr};
47 0 : CHK_RET(BuildCcuConnection(ccuConnectionInfo, ccuConnection));
48 :
49 0 : ccuTransport.reset(new (std::nothrow) CcuTransport(socket, std::move(ccuConnection), cclBufferInfo));
50 0 : CHK_PTR_NULL(ccuTransport);
51 0 : CHK_RET(ccuTransport->Init());
52 :
53 0 : return HcclResult::HCCL_SUCCESS;
54 0 : }
55 :
56 15 : HcclResult CcuCreateTransport(
57 : Hccl::Socket* socket, const CcuTransport::CcuConnectionInfo& ccuConnectionInfo,
58 : const std::vector<CcuTransport::CclBufferInfo>& bufferInfos, std::unique_ptr<CcuTransport>& ccuTransport)
59 : {
60 15 : CHK_PTR_NULL(socket);
61 15 : std::unique_ptr<CcuConnection> ccuConnection{nullptr};
62 15 : CHK_RET(BuildCcuConnection(ccuConnectionInfo, ccuConnection));
63 :
64 15 : if (bufferInfos.size() == 0) {
65 0 : HCCL_ERROR("[CcuCreateTransport] bufferNum is 0.");
66 0 : return HCCL_E_PARA;
67 : }
68 15 : ccuTransport.reset(new (std::nothrow) CcuTransport(socket, std::move(ccuConnection), bufferInfos));
69 15 : CHK_PTR_NULL(ccuTransport);
70 : // 可能申请xn cke失败，需要回退
71 15 : auto ret = ccuTransport->Init();
72 15 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
73 0 : HCCL_WARNING("[%s] ccuTransport init failed, ccu transport resources unavailable.", __func__);
74 0 : return ret;
75 : }
76 15 : CHK_RET(ret);
77 :
78 15 : return HcclResult::HCCL_SUCCESS;
79 15 : }
80 :
81 45 : CcuTransport::CcuTransport(
82 45 : Hccl::Socket* socket, std::unique_ptr<CcuConnection>&& connection, const CclBufferInfo& locCclBufInfo)
83 45 : : socket_(socket),
84 135 : ccuConnection_(std::move(connection))
85 : {
86 45 : locBufferInfos_.push_back(locCclBufInfo);
87 45 : }
88 :
89 17 : CcuTransport::CcuTransport(
90 17 : Hccl::Socket* socket, std::unique_ptr<CcuConnection>&& connection, const std::vector<CclBufferInfo>& bufferInfos)
91 17 : : socket_(socket),
92 17 : ccuConnection_(std::move(connection)),
93 68 : locBufferInfos_(bufferInfos)
94 17 : {}
95 :
96 15 : HcclResult CcuTransport::Init()
97 : {
98 15 : dieId_ = ccuConnection_->GetDieId();
99 15 : devLogicId_ = ccuConnection_->GetDevLogicId();
100 15 : auto ret = AppendCkes(INIT_CKE_NUM);
101 15 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
102 0 : return ret;
103 : }
104 15 : CHK_RET(ret);
105 :
106 15 : ret = AppendXns(INIT_XN_NUM);
107 15 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
108 0 : return ret;
109 : }
110 15 : CHK_RET(ret);
111 :
112 15 : transStatus_ = TransStatus::INIT;
113 : // 正常建链路径在 Init 成功时确认本端资源充足；msg-only(资源不足)路径由
114 : // ConstructMsgOnlyTransport 单独覆盖为 UNAVAIL/FAILED，避免依赖调用方设置
115 15 : locResStatus_ = CcuResStatus::RES_OK;
116 15 : return HCCL_SUCCESS;
117 : }
118 :
119 12 : CcuTransport::TransStatus CcuTransport::GetStatus()
120 : {
121 21 : if (transStatus_ == TransStatus::READY || transStatus_ == TransStatus::CONNECT_FAILED
122 21 : || transStatus_ == TransStatus::SOCKET_TIMEOUT) {
123 11 : return transStatus_;
124 : }
125 :
126 1 : if (StatusMachine() != HcclResult::HCCL_SUCCESS) {
127 0 : HCCL_ERROR("[CcuTransport][%s] failed, %s.", __func__, transStatus_.Describe().c_str());
128 0 : transStatus_ = TransStatus::CONNECT_FAILED;
129 : }
130 :
131 1 : return transStatus_;
132 : }
133 :
134 15 : HcclResult CcuTransport::AppendCkes(uint32_t ckesNum)
135 : {
136 15 : std::vector<ResInfo> resInfo;
137 15 : auto ret = CcuDevMgrImp::AllocCke(devLogicId_, dieId_, ckesNum, resInfo);
138 15 : CHK_PRT_RET(
139 : ret == HcclResult::HCCL_E_UNAVAIL,
140 : HCCL_WARNING("[CcuTransport][%s] failed, the resource is not enough.", __func__), ret);
141 15 : CHK_RET(ret);
142 :
143 15 : const uint32_t resSize = resInfo.size();
144 30 : for (uint32_t i = 0; i < resSize; i++) {
145 15 : const uint32_t ckeNum = resInfo[i].num;
146 15 : const uint32_t ckesSartId = resInfo[i].startId;
147 75 : for (uint32_t j = 0; j < ckeNum; j++) {
148 60 : locRes_.ckes.emplace_back(ckesSartId + j);
149 : }
150 : }
151 15 : ckesRes_.push_back(resInfo);
152 15 : return HCCL_SUCCESS;
153 15 : }
154 :
155 15 : HcclResult CcuTransport::AppendXns(uint32_t xnsNum)
156 : {
157 15 : std::vector<ResInfo> resInfo;
158 15 : auto ret = CcuDevMgrImp::AllocXn(devLogicId_, dieId_, xnsNum, resInfo);
159 15 : CHK_PRT_RET(
160 : ret == HcclResult::HCCL_E_UNAVAIL,
161 : HCCL_WARNING("[CcuTransport][%s] failed, the resource is not enough.", __func__), ret);
162 15 : CHK_RET(ret);
163 :
164 15 : const uint32_t resSize = resInfo.size();
165 30 : for (uint32_t i = 0; i < resSize; i++) {
166 15 : uint32_t xnNum = resInfo[i].num;
167 15 : uint32_t xnsSartId = resInfo[i].startId;
168 75 : for (uint32_t j = 0; j < xnNum; j++) {
169 60 : locRes_.xns.emplace_back(xnsSartId + j);
170 : }
171 : }
172 15 : xnsRes_.push_back(resInfo);
173 15 : return HCCL_SUCCESS;
174 15 : }
175 :
176 1 : HcclResult CcuTransport::AppendCntXns()
177 : {
178 3 : for (auto& cntXns : locRes_.cntXns) {
179 2 : if (cntXns.second == INVALID_UINT) {
180 2 : uint32_t wishCntXnId = 0;
181 2 : auto ret = CcuDevMgrImp::AllocWishCntXn(devLogicId_, dieId_, cntXns.first, wishCntXnId);
182 2 : CHK_PRT_RET(
183 : ret == HcclResult::HCCL_E_UNAVAIL,
184 : HCCL_ERROR("[CcuTransport][%s] failed, the resource is not enough.", __func__), ret);
185 2 : CHK_RET(ret);
186 2 : cntXns.second = wishCntXnId;
187 : }
188 : }
189 1 : return HCCL_SUCCESS;
190 : }
191 :
192 2 : HcclResult CcuTransport::StatusMachine()
193 : {
194 : EXCEPTION_HANDLE_BEGIN
195 2 : Hccl::SocketStatus socketStatus = socket_->GetAsyncStatus();
196 2 : if (socketStatus == Hccl::SocketStatus::INIT || socketStatus == Hccl::SocketStatus::TIMEOUT) {
197 0 : HCCL_ERROR("[CcuTransport][GetStatus] socket timeout or no link, please check");
198 0 : return HcclResult::HCCL_E_INTERNAL;
199 : }
200 :
201 2 : if (socketStatus != Hccl::SocketStatus::OK) {
202 0 : return HcclResult::HCCL_SUCCESS; // 操作成功，保持当前状态
203 : }
204 0 : EXCEPTION_HANDLE_END
205 :
206 2 : switch (transStatus_) {
207 1 : case CcuTransport::TransStatus::INIT: {
208 1 : if (locResStatus_ != CcuResStatus::RES_OK) {
209 : // 本端资源不足或资源申请失败, 无ccuConnection_, 直接发状态标记
210 1 : CHK_RET(SendDataSize());
211 1 : transStatus_ = TransStatus::SEND_DATA_SIZE;
212 1 : break;
213 : }
214 0 : auto connStatus = ccuConnection_->GetStatus();
215 0 : if (connStatus == CcuConnStatus::CONN_INVALID) {
216 0 : HCCL_ERROR(
217 : "[CcuTransport][GetStatus] connection status[%s] failed."
218 : " please check.",
219 : connStatus.Describe().c_str());
220 0 : return HcclResult::HCCL_E_INTERNAL;
221 : }
222 :
223 0 : if (connStatus == CcuConnStatus::EXCHANGEABLE || connStatus == CcuConnStatus::CONNECTED) {
224 : // connection完成本端资源创建或复用时，发送本端资源信息
225 0 : CHK_RET(SendDataSize());
226 0 : transStatus_ = TransStatus::SEND_DATA_SIZE;
227 : }
228 :
229 : // connection状态非错误但未达到目标状态时，transport保持当前状态
230 0 : break;
231 : }
232 0 : case CcuTransport::TransStatus::SEND_DATA_SIZE:
233 0 : CHK_RET(RecvDataSize());
234 0 : transStatus_ = TransStatus::RECV_DATA_SIZE;
235 0 : break;
236 0 : case CcuTransport::TransStatus::RECV_DATA_SIZE:
237 0 : CHK_RET(SendConnAndTransInfo());
238 0 : transStatus_ = TransStatus::SEND_ALL_INFO;
239 0 : break;
240 0 : case CcuTransport::TransStatus::SEND_ALL_INFO:
241 0 : CHK_RET(RecvConnAndTransInfo());
242 0 : transStatus_ = TransStatus::RECV_ALL_INFO;
243 0 : break;
244 0 : case CcuTransport::TransStatus::RECV_ALL_INFO:
245 0 : CHK_RET(RecvDataProcess());
246 0 : if (locResStatus_ == CcuResStatus::RES_OK && rmtResStatus_ == CcuResStatus::RES_OK) {
247 0 : CHK_RET(ccuConnection_->ImportJetty());
248 : }
249 0 : transStatus_ = TransStatus::SEND_FIN;
250 0 : break;
251 0 : case CcuTransport::TransStatus::SEND_FIN: {
252 0 : if (locResStatus_ != CcuResStatus::RES_OK || rmtResStatus_ != CcuResStatus::RES_OK) {
253 0 : CHK_RET(SendFinish());
254 0 : transStatus_ = CcuTransport::TransStatus::RECVING_FIN;
255 0 : break;
256 : }
257 0 : auto connStatus = ccuConnection_->GetStatus();
258 0 : if (connStatus == CcuConnStatus::CONN_INVALID) {
259 0 : HCCL_ERROR(
260 : "[CcuTransport][GetStatus] connection status[%s] failed."
261 : " please check",
262 : connStatus.Describe().c_str());
263 0 : return HcclResult::HCCL_E_INTERNAL;
264 : }
265 :
266 0 : if (connStatus == CcuConnStatus::CONNECTED) {
267 0 : CHK_RET(SendFinish());
268 0 : transStatus_ = CcuTransport::TransStatus::RECVING_FIN;
269 : }
270 0 : break;
271 : }
272 0 : case CcuTransport::TransStatus::RECVING_FIN:
273 0 : CHK_RET(RecvFinish());
274 0 : transStatus_ = CcuTransport::TransStatus::RECV_FIN;
275 0 : break;
276 1 : case CcuTransport::TransStatus::RECV_FIN:
277 1 : CHK_RET(CheckFinish());
278 1 : if (locResStatus_ != CcuResStatus::RES_OK || rmtResStatus_ != CcuResStatus::RES_OK) {
279 1 : HCCL_WARNING(
280 : "[CcuTransport][%s] resource status not OK after FIN (loc[%u], rmt[%u]) set CONNECT_FAILED",
281 : __func__, static_cast<uint8_t>(locResStatus_), static_cast<uint8_t>(rmtResStatus_));
282 1 : transStatus_ = TransStatus::CONNECT_FAILED;
283 1 : break;
284 : }
285 0 : transStatus_ = CcuTransport::TransStatus::READY;
286 0 : break;
287 0 : case CcuTransport::TransStatus::SEND_TRANS_RES:
288 0 : CHK_RET(SendTransInfo());
289 0 : transStatus_ = CcuTransport::TransStatus::RECVING_TRANS_RES;
290 0 : break;
291 0 : case CcuTransport::TransStatus::RECVING_TRANS_RES:
292 0 : CHK_RET(RecvTransInfo());
293 0 : transStatus_ = CcuTransport::TransStatus::RECV_TRANS_RES;
294 0 : break;
295 0 : case CcuTransport::TransStatus::RECV_TRANS_RES:
296 0 : CHK_RET(RecvTransInfoProcess());
297 0 : transStatus_ = CcuTransport::TransStatus::SEND_FIN;
298 0 : break;
299 0 : default:
300 0 : HCCL_ERROR("[CcuTransport][%s] failed, error status[%s].", __func__, transStatus_.Describe().c_str());
301 0 : transStatus_ = CcuTransport::TransStatus::CONNECT_FAILED;
302 0 : break;
303 : }
304 2 : return HcclResult::HCCL_SUCCESS;
305 : }
306 :
307 3 : HcclResult CcuTransport::SendDataSize()
308 : {
309 3 : Hccl::BinaryStream binaryStream;
310 : // 所有路径先写 locResStatus (int), 接收端据此分流
311 3 : int locResStatus = static_cast<int>(locResStatus_);
312 3 : binaryStream << locResStatus;
313 3 : if (locResStatus_ == CcuResStatus::RES_OK) {
314 0 : CHK_RET(HandshakeMsgPack(binaryStream));
315 0 : CHK_RET(ConnInfoPack(binaryStream));
316 0 : CHK_RET(TransResPack(binaryStream));
317 0 : CHK_RET(BufferInfoPack(binaryStream, locBufferInfos_));
318 : }
319 3 : binaryStream.Dump(sendData_);
320 3 : u32 sendSize = sendData_.size();
321 :
322 : // 发送数据包尺寸
323 : EXCEPTION_HANDLE_BEGIN
324 3 : socket_->SendAsync(&sendSize, sizeof(sendSize));
325 0 : EXCEPTION_HANDLE_END
326 3 : HCCL_INFO(
327 : "[CcuTransport::%s] Send size[%u] of data success. [%zu] bytes sent. locResStatus[%u]", __func__, sendSize,
328 : sizeof(sendSize), static_cast<uint8_t>(locResStatus_));
329 3 : return HcclResult::HCCL_SUCCESS;
330 3 : }
331 :
332 0 : HcclResult CcuTransport::RecvDataSize()
333 : {
334 : // 接收数据包尺寸
335 : EXCEPTION_HANDLE_BEGIN
336 0 : socket_->RecvAsync(reinterpret_cast<u8*>(&exchangeDataSize_), sizeof(exchangeDataSize_));
337 0 : EXCEPTION_HANDLE_END
338 0 : HCCL_INFO(
339 : "[CcuTransport::%s] Receive size[%u] of data success. [%zu] bytes received.", __func__, exchangeDataSize_,
340 : sizeof(exchangeDataSize_));
341 0 : return HcclResult::HCCL_SUCCESS;
342 : }
343 :
344 0 : HcclResult CcuTransport::SendConnAndTransInfo()
345 : {
346 : // 当前socket失败会抛异常，需要统一整改
347 : EXCEPTION_HANDLE_BEGIN
348 0 : socket_->SendAsync(sendData_.data(), sendData_.size());
349 0 : EXCEPTION_HANDLE_END
350 0 : return HcclResult::HCCL_SUCCESS;
351 : }
352 :
353 0 : HcclResult CcuTransport::RecvConnAndTransInfo()
354 : {
355 0 : recvData_.resize(exchangeDataSize_);
356 : EXCEPTION_HANDLE_BEGIN
357 0 : socket_->RecvAsync(reinterpret_cast<u8*>(recvData_.data()), recvData_.size());
358 0 : EXCEPTION_HANDLE_END
359 0 : return HcclResult::HCCL_SUCCESS;
360 : }
361 :
362 4 : HcclResult CcuTransport::RecvDataProcess()
363 : {
364 4 : if (exchangeDataSize_ < sizeof(int)) {
365 1 : HCCL_ERROR(
366 : "[CcuTransport][%s] exchangeDataSize[%u] < sizeof(int)[%zu], protocol mismatch.", __func__,
367 : exchangeDataSize_, sizeof(int));
368 1 : return HcclResult::HCCL_E_INTERNAL;
369 : }
370 3 : Hccl::BinaryStream binaryStream(recvData_);
371 3 : int rmtResStatusInt = static_cast<int>(CcuResStatus::RES_UNKNOWN);
372 3 : binaryStream >> rmtResStatusInt;
373 : // 校验对端传值范围：越界值强转后比较行为未定义，按协议错误处理
374 3 : if (rmtResStatusInt < static_cast<int>(CcuResStatus::RES_UNKNOWN)
375 3 : || rmtResStatusInt > static_cast<int>(CcuResStatus::RES_FAILED)) {
376 1 : HCCL_ERROR(
377 : "[CcuTransport][%s] invalid rmtResStatus[%d], out of range[%d,%d].", __func__, rmtResStatusInt,
378 : static_cast<int>(CcuResStatus::RES_UNKNOWN), static_cast<int>(CcuResStatus::RES_FAILED));
379 1 : return HcclResult::HCCL_E_INTERNAL;
380 : }
381 2 : rmtResStatus_ = static_cast<CcuResStatus>(rmtResStatusInt);
382 :
383 2 : if (rmtResStatus_ != CcuResStatus::RES_OK) {
384 2 : HCCL_WARNING(
385 : "[CcuTransport][%s] remote resource status[%u], skip rest unpack.", __func__,
386 : static_cast<uint8_t>(rmtResStatus_));
387 2 : return HcclResult::HCCL_SUCCESS;
388 : }
389 0 : if (locResStatus_ != CcuResStatus::RES_OK) {
390 0 : HCCL_WARNING(
391 : "[CcuTransport][%s] loc resource status[%u], skip remote full unpack.", __func__,
392 : static_cast<uint8_t>(locResStatus_));
393 0 : return HcclResult::HCCL_SUCCESS;
394 : }
395 0 : CHK_RET(HandshakeMsgUnpack(binaryStream));
396 0 : CHK_RET(ConnInfoUnpackProc(binaryStream));
397 0 : CHK_RET(TransResUnpackProc(binaryStream));
398 0 : rmtBufferVec_.clear();
399 0 : CHK_RET(BufferInfoUnpack(binaryStream));
400 0 : return HcclResult::HCCL_SUCCESS;
401 3 : }
402 :
403 0 : HcclResult CcuTransport::SendTransInfo()
404 : {
405 0 : Hccl::BinaryStream binaryStream;
406 0 : TransResPack(binaryStream);
407 0 : binaryStream.Dump(sendTrans_);
408 : EXCEPTION_HANDLE_BEGIN
409 0 : socket_->SendAsync(sendTrans_.data(), sendTrans_.size());
410 0 : EXCEPTION_HANDLE_END
411 0 : exchangeDataSize_ = sendTrans_.size();
412 0 : return HcclResult::HCCL_SUCCESS;
413 0 : }
414 :
415 0 : HcclResult CcuTransport::RecvTransInfo()
416 : {
417 0 : recvTrans_.resize(exchangeDataSize_);
418 : EXCEPTION_HANDLE_BEGIN
419 0 : socket_->RecvAsync(reinterpret_cast<u8*>(recvTrans_.data()), recvTrans_.size());
420 0 : EXCEPTION_HANDLE_END
421 0 : return HcclResult::HCCL_SUCCESS;
422 : }
423 :
424 0 : HcclResult CcuTransport::RecvTransInfoProcess()
425 : {
426 0 : Hccl::BinaryStream binaryStream(recvTrans_);
427 0 : TransResUnpackProc(binaryStream);
428 0 : return HcclResult::HCCL_SUCCESS;
429 0 : }
430 :
431 1 : HcclResult CcuTransport::HandshakeMsgPack(Hccl::BinaryStream& binaryStream)
432 : {
433 1 : binaryStream << attr_.handshakeMsg;
434 1 : HCCL_INFO(
435 : "[CcuTransport][%s] start pack handshakeMsg, attr.handshakeMsg.size[%zu]", __func__, attr_.handshakeMsg.size());
436 1 : return HcclResult::HCCL_SUCCESS;
437 : }
438 :
439 0 : HcclResult CcuTransport::ConnInfoPack(Hccl::BinaryStream& binaryStream) const
440 : {
441 0 : std::vector<char> dtoData{};
442 0 : CHK_RET(ccuConnection_->Serialize(dtoData));
443 0 : binaryStream << dtoData;
444 0 : HCCL_INFO("[CcuTransport][%s] start pack connInfo, dtoData.size[%zu]", __func__, dtoData.size());
445 0 : return HcclResult::HCCL_SUCCESS;
446 0 : }
447 :
448 0 : HcclResult CcuTransport::TransResPack(Hccl::BinaryStream& binaryStream)
449 : {
450 0 : const uint32_t locCkesSize = locRes_.ckes.size();
451 0 : binaryStream << locCkesSize;
452 0 : const uint32_t locCkeSize = locRes_.ckes.size();
453 0 : for (uint32_t i = 0; i < locCkeSize; i++) {
454 0 : binaryStream << locRes_.ckes[i];
455 : }
456 :
457 0 : const uint32_t locXnsSize = locRes_.xns.size();
458 0 : binaryStream << locXnsSize;
459 0 : for (uint32_t i = 0; i < locRes_.xns.size(); i++) {
460 0 : binaryStream << locRes_.xns[i];
461 : }
462 :
463 0 : HCCL_INFO("Send ckesSize[%u], xnsSize[%u]", locCkesSize, locXnsSize);
464 0 : return HcclResult::HCCL_SUCCESS;
465 : }
466 :
467 0 : HcclResult CcuTransport::TransCntXnResPack(Hccl::BinaryStream& binaryStream)
468 : {
469 0 : const uint32_t locCntXnsSize = locRes_.cntXns.size();
470 0 : binaryStream << locCntXnsSize;
471 0 : for (auto& cntXns : locRes_.cntXns) {
472 0 : binaryStream << cntXns.first << cntXns.second;
473 0 : HCCL_INFO("Send resGroupTag[%s], wishCntXn[%u]", cntXns.first.c_str(), cntXns.second);
474 : }
475 :
476 0 : return HcclResult::HCCL_SUCCESS;
477 : }
478 :
479 0 : HcclResult CcuTransport::BufferInfoPack(Hccl::BinaryStream& binaryStream, std::vector<CclBufferInfo>& bufferVec) const
480 : {
481 0 : u32 locBufferNum = bufferVec.size();
482 0 : binaryStream << locBufferNum;
483 0 : for (u32 pos = 0; pos < locBufferNum; ++pos) {
484 0 : bufferVec[pos].Pack(binaryStream);
485 : }
486 0 : return HcclResult::HCCL_SUCCESS;
487 : }
488 :
489 1 : HcclResult CcuTransport::HandshakeMsgUnpack(Hccl::BinaryStream& binaryStream)
490 : {
491 1 : binaryStream >> rmtHandshakeMsg_;
492 :
493 1 : if (attr_.handshakeMsg.size() != rmtHandshakeMsg_.size()) {
494 0 : HCCL_ERROR("handshakeMsg size=%zu is not equal to rmt=%zu", attr_.handshakeMsg.size(), rmtHandshakeMsg_.size());
495 0 : return HcclResult::HCCL_E_INTERNAL;
496 : }
497 1 : HCCL_INFO("[CcuTransport][%s] unpack handshakeMsg, rmtHandshakeMsg.size[%zu]", __func__, rmtHandshakeMsg_.size());
498 1 : return HcclResult::HCCL_SUCCESS;
499 : }
500 :
501 0 : HcclResult CcuTransport::ConnInfoUnpackProc(Hccl::BinaryStream& binaryStream) const
502 : {
503 0 : std::vector<char> dtoData{};
504 0 : binaryStream >> dtoData;
505 0 : CHK_RET(ccuConnection_->Deserialize(dtoData));
506 0 : HCCL_INFO("[CcuTransport][%s] start unpack connInfo, dtoData.size[%zu]", __func__, dtoData.size());
507 0 : return HcclResult::HCCL_SUCCESS;
508 0 : }
509 :
510 0 : HcclResult CcuTransport::TransResUnpackProc(Hccl::BinaryStream& binaryStream)
511 : {
512 0 : uint32_t resSize{0};
513 0 : binaryStream >> resSize;
514 0 : rmtRes_.ckes.clear();
515 0 : for (uint32_t i = 0; i < resSize; i++) {
516 0 : uint32_t cke{0};
517 0 : binaryStream >> cke;
518 0 : rmtRes_.ckes.push_back(cke);
519 : }
520 0 : HCCL_INFO("Recv ckesSize[%u]", resSize);
521 :
522 0 : binaryStream >> resSize;
523 0 : rmtRes_.xns.clear();
524 0 : for (uint32_t i = 0; i < resSize; i++) {
525 0 : uint32_t xn{0};
526 0 : binaryStream >> xn;
527 0 : rmtRes_.xns.push_back(xn);
528 : }
529 0 : HCCL_INFO("Recv xnsSize[%u]", resSize);
530 :
531 0 : return HcclResult::HCCL_SUCCESS;
532 : }
533 :
534 0 : HcclResult CcuTransport::TransCntXnResUnpackProc(Hccl::BinaryStream& binaryStream)
535 : {
536 0 : uint32_t resSize{0};
537 0 : binaryStream >> resSize;
538 0 : HCCL_INFO("Recv resGroupTagSize[%u]", resSize);
539 0 : for (uint32_t num = 0; num < resSize; num++) {
540 0 : std::string resGroupTag;
541 0 : uint32_t wishCntXn = 0;
542 0 : binaryStream >> resGroupTag >> wishCntXn;
543 0 : HCCL_INFO(
544 : "Recv resGroupTag[%s], wishCntXn[%u], locRes_.cntXns size[%zu]", resGroupTag.c_str(), wishCntXn,
545 : locRes_.cntXns.size());
546 0 : if (locRes_.cntXns.find(resGroupTag) == locRes_.cntXns.end()) {
547 0 : HCCL_ERROR("Recv resGroupTag[%s] not in locRes.", resGroupTag.c_str());
548 0 : return HcclResult::HCCL_E_INTERNAL;
549 : }
550 0 : auto iter = rmtRes_.cntXns.find(resGroupTag);
551 0 : if (iter == rmtRes_.cntXns.end()) {
552 0 : rmtRes_.cntXns.insert(std::make_pair(resGroupTag, wishCntXn));
553 0 : } else if (iter->second != wishCntXn) {
554 0 : HCCL_ERROR(
555 : "Recv resGroupTag[%s], current rmt cnt xn[%u] is not equal to recv cnt xn[%u].", resGroupTag.c_str(),
556 : iter->second, wishCntXn);
557 0 : return HcclResult::HCCL_E_INTERNAL;
558 : }
559 0 : }
560 0 : return HcclResult::HCCL_SUCCESS;
561 : }
562 :
563 0 : HcclResult CcuTransport::BufferInfoUnpack(Hccl::BinaryStream& binaryStream)
564 : {
565 0 : u32 rmtBufferNum{0};
566 0 : binaryStream >> rmtBufferNum;
567 0 : CHK_PRT_RET(
568 : rmtBufferNum == 0 || rmtBufferNum > MAX_BUFFER_NUM,
569 : HCCL_ERROR(
570 : "[CcuTransport][BufferInfoUnpack] rmtBufferNum[%u] is zero or exceeds limit[%u]", rmtBufferNum,
571 : MAX_BUFFER_NUM),
572 : HCCL_E_PARA);
573 0 : HCCL_INFO("[CcuTransport][BufferInfoUnpack] rmtBufferNum[%u]", rmtBufferNum);
574 0 : for (u32 pos = 0; pos < rmtBufferNum; ++pos) {
575 0 : CclBufferInfo rmtBufferInfo{};
576 0 : rmtBufferInfo.Unpack(binaryStream);
577 0 : std::string memInfo(rmtBufferInfo.memInfo.data(), strnlen(rmtBufferInfo.memInfo.data(), HCCL_RES_TAG_MAX_LEN));
578 0 : if (memInfo == "HcclBuffer") {
579 0 : rmtHcclBufferInfo_ = rmtBufferInfo;
580 : }
581 0 : rmtBufferVec_.push_back(std::make_unique<Hccl::RemoteUbRmaBuffer>(
582 0 : reinterpret_cast<uintptr_t>(rmtBufferInfo.addr), rmtBufferInfo.size, rmtBufferInfo.tokenId,
583 0 : rmtBufferInfo.tokenValue, Hccl::CommMemTypeToHcclMemType(rmtBufferInfo.type), memInfo));
584 0 : }
585 0 : return HcclResult::HCCL_SUCCESS;
586 : }
587 :
588 0 : HcclResult CcuTransport::SendFinish()
589 : {
590 0 : sendFinishMsg_ = std::vector<char>(FINISH_MSG, FINISH_MSG + FINISH_MSG_SIZE);
591 : EXCEPTION_HANDLE_BEGIN
592 0 : socket_->SendAsync(sendFinishMsg_.data(), FINISH_MSG_SIZE);
593 0 : EXCEPTION_HANDLE_END
594 0 : return HcclResult::HCCL_SUCCESS;
595 : }
596 :
597 0 : HcclResult CcuTransport::RecvFinish()
598 : {
599 0 : recvFinishMsg_.resize(FINISH_MSG_SIZE);
600 : EXCEPTION_HANDLE_BEGIN
601 0 : socket_->RecvAsync(reinterpret_cast<u8*>(recvFinishMsg_.data()), FINISH_MSG_SIZE);
602 0 : EXCEPTION_HANDLE_END
603 0 : return HcclResult::HCCL_SUCCESS;
604 : }
605 :
606 3 : HcclResult CcuTransport::CheckFinish()
607 : {
608 6 : const std::string sendFinishMsgStr(sendFinishMsg_.begin(), sendFinishMsg_.end());
609 3 : const std::string recvFinishMsgStr(recvFinishMsg_.begin(), recvFinishMsg_.end());
610 3 : if (sendFinishMsgStr != recvFinishMsgStr) {
611 1 : HCCL_ERROR(
612 : "[CcuTransport][RecvFinish]msgRecv[%s] and msgSend[%s] are not equal", recvFinishMsgStr.c_str(),
613 : sendFinishMsgStr.c_str());
614 1 : return HcclResult::HCCL_E_INTERNAL;
615 : }
616 :
617 2 : return HcclResult::HCCL_SUCCESS;
618 3 : }
619 :
620 62 : HcclResult CcuTransport::ReleaseTransRes()
621 : {
622 77 : for (uint32_t i = 0; i < ckesRes_.size(); i++) {
623 15 : if (ckesRes_[i].empty()) {
624 0 : continue;
625 : }
626 15 : auto ret = CcuDevMgrImp::ReleaseCke(devLogicId_, dieId_, ckesRes_[i]);
627 15 : if (ret != HcclResult::HCCL_SUCCESS) {
628 0 : HCCL_ERROR(
629 : "[CcuTransport][%s] release ckes failed but passed, "
630 : "devLogicId[%d] dieId[%u].",
631 : __func__, devLogicId_, dieId_);
632 : }
633 : }
634 62 : ckesRes_.clear();
635 :
636 77 : for (uint32_t i = 0; i < xnsRes_.size(); i++) {
637 15 : if (xnsRes_[i].empty()) {
638 0 : continue;
639 : }
640 15 : auto ret = CcuDevMgrImp::ReleaseXn(devLogicId_, dieId_, xnsRes_[i]);
641 15 : if (ret != HcclResult::HCCL_SUCCESS) {
642 0 : HCCL_ERROR(
643 : "[CcuTransport][%s] release xns failed but passed, "
644 : "devLogicId[%d] dieId[%u].",
645 : __func__, devLogicId_, dieId_);
646 : }
647 : }
648 62 : xnsRes_.clear();
649 :
650 62 : return HcclResult::HCCL_SUCCESS;
651 : }
652 :
653 18 : uint32_t CcuTransport::GetDieId() const { return dieId_; }
654 :
655 33 : uint32_t CcuTransport::GetChannelId() const
656 : {
657 : // msg-only(资源不足)transport 无 ccuConnection_，返回无效 id，避免空指针解引用
658 33 : if (ccuConnection_ == nullptr) {
659 0 : return UINT32_MAX;
660 : }
661 33 : return ccuConnection_->GetChannelId();
662 : }
663 :
664 15 : HcclResult CcuTransport::GetLocCkeByIndex(const uint32_t index, uint32_t& locCkeId) const
665 : {
666 15 : CHK_PRT_RET(
667 : locRes_.ckes.empty(), HCCL_ERROR("[CcuTransport][%s] failed, local resources is empty.", __func__),
668 : HcclResult::HCCL_E_PARA);
669 :
670 14 : CHK_PRT_RET(
671 : index >= locRes_.ckes.size(),
672 : HCCL_ERROR(
673 : "[CcuTransport][%s] failed, index[%u] is larger than size[%u].", __func__, index, locRes_.ckes.size()),
674 : HcclResult::HCCL_E_PARA);
675 :
676 13 : locCkeId = locRes_.ckes[index];
677 13 : return HcclResult::HCCL_SUCCESS;
678 : }
679 :
680 13 : HcclResult CcuTransport::GetLocXnByIndex(const uint32_t index, uint32_t& locXnId) const
681 : {
682 13 : CHK_PRT_RET(
683 : locRes_.xns.empty(), HCCL_ERROR("[CcuTransport][%s] failed, local resources is empty.", __func__),
684 : HcclResult::HCCL_E_PARA);
685 :
686 12 : CHK_PRT_RET(
687 : index >= locRes_.xns.size(),
688 : HCCL_ERROR(
689 : "[CcuTransport][%s] failed, index[%u] is larger than size[%u].", __func__, index, locRes_.xns.size()),
690 : HcclResult::HCCL_E_PARA);
691 :
692 11 : locXnId = locRes_.xns[index];
693 11 : return HcclResult::HCCL_SUCCESS;
694 : }
695 :
696 2 : HcclResult CcuTransport::GetRmtCkeByIndex(const uint32_t index, uint32_t& rmtCkeId) const
697 : {
698 2 : CHK_PRT_RET(
699 : rmtRes_.ckes.empty(), HCCL_ERROR("[CcuTransport][%s] failed, local resources is empty.", __func__),
700 : HcclResult::HCCL_E_PARA);
701 :
702 1 : CHK_PRT_RET(
703 : index >= rmtRes_.ckes.size(),
704 : HCCL_ERROR(
705 : "[CcuTransport][%s] failed, index[%u] is larger than size[%u].", __func__, index, rmtRes_.ckes.size()),
706 : HcclResult::HCCL_E_PARA);
707 :
708 1 : rmtCkeId = rmtRes_.ckes[index];
709 1 : return HcclResult::HCCL_SUCCESS;
710 : }
711 :
712 3 : HcclResult CcuTransport::GetRmtXnByIndex(const uint32_t index, uint32_t& rmtXnId) const
713 : {
714 3 : CHK_PRT_RET(
715 : rmtRes_.xns.empty(), HCCL_ERROR("[CcuTransport][%s] failed, local resources is empty.", __func__),
716 : HcclResult::HCCL_E_PARA);
717 :
718 2 : CHK_PRT_RET(
719 : index >= rmtRes_.xns.size(),
720 : HCCL_ERROR(
721 : "[CcuTransport][%s] failed, index[%u] is larger than size[%u].", __func__, index, rmtRes_.xns.size()),
722 : HcclResult::HCCL_E_PARA);
723 :
724 1 : rmtXnId = rmtRes_.xns[index];
725 1 : return HcclResult::HCCL_SUCCESS;
726 : }
727 :
728 1 : HcclResult CcuTransport::GetRmtWishCntXnAddr(const std::string& resGroupTag, uint64_t& wishCntXnAddr) const
729 : {
730 1 : auto iter = rmtRes_.cntXns.find(resGroupTag);
731 1 : if (iter == rmtRes_.cntXns.end()) {
732 1 : HCCL_ERROR("[CcuTransport][%s] failed, resGroupTag[%s] is not found.", __func__, resGroupTag.c_str());
733 1 : return HCCL_E_NOT_FOUND;
734 : }
735 :
736 0 : const uint32_t wishCntXn = iter->second;
737 0 : CHK_RET(GetRmtVarAddrByXnId(wishCntXn, wishCntXnAddr));
738 0 : HCCL_DEBUG(
739 : "[CcuTransport][%s] resGroupTag[%s], wishCntXnAddr[%u][0x%llx]", __func__, resGroupTag.c_str(), wishCntXn,
740 : wishCntXnAddr);
741 0 : return HCCL_SUCCESS;
742 : }
743 :
744 1 : HcclResult CcuTransport::GetLocBuffer(CclBufferInfo& bufferInfo, const uint32_t& bufNum) const
745 : {
746 : (void)bufNum;
747 1 : bufferInfo = locBufferInfos_[0];
748 1 : return HCCL_SUCCESS;
749 : }
750 :
751 1 : HcclResult CcuTransport::GetRmtBuffer(CclBufferInfo& bufferInfo, const uint32_t& bufNum) const
752 : {
753 : (void)bufNum;
754 1 : bufferInfo = rmtHcclBufferInfo_;
755 1 : return HCCL_SUCCESS;
756 : }
757 :
758 1 : HcclResult CcuTransport::GetCkeNum(uint32_t& ckeNum) const
759 : {
760 1 : ckeNum = locRes_.ckes.size();
761 1 : return HcclResult::HCCL_SUCCESS;
762 : }
763 :
764 0 : HcclResult CcuTransport::GetRmtSignalAddrByIndex(uint32_t index, uint64_t& rmtCkeAddr) const
765 : {
766 0 : uint32_t rmtCkeId{0};
767 0 : uint64_t ckeOffsetCcumAddr{0};
768 0 : CHK_RET(GetRmtCkeByIndex(index, rmtCkeId));
769 0 : CHK_PRT_RET(
770 : CcuDevMgrImp::GetCkeOffsetCcumAddrById(devLogicId_, dieId_, rmtCkeId, ckeOffsetCcumAddr),
771 : HCCL_ERROR(
772 : "[CcuTransport][%s] Failed to get cke offset address. devLogicId = %d, dieId = %u.", __func__, devLogicId_,
773 : dieId_),
774 : HCCL_E_INTERNAL);
775 0 : uint64_t rmtResourceAddr = ccuConnection_->GetRmtCcuBufAddr();
776 0 : HCCL_DEBUG(
777 : "[CcuTransport][%s] index[%u] rmtCcuBufAddr[0x%llx], ckeAddr[%u][0x%llx]", __func__, index, rmtResourceAddr,
778 : rmtCkeId, ckeOffsetCcumAddr);
779 0 : if (ckeOffsetCcumAddr > UINT64_MAX - rmtResourceAddr) {
780 0 : HCCL_ERROR(
781 : "[CcuTransport][%s] failed, rmtResourceAddr[%llx] + ckeOffsetCcumAddr[%llx] is overflow.", __func__,
782 : rmtResourceAddr, ckeOffsetCcumAddr);
783 0 : return HcclResult::HCCL_E_INTERNAL;
784 : }
785 0 : rmtCkeAddr = rmtResourceAddr + ckeOffsetCcumAddr;
786 0 : return HCCL_SUCCESS;
787 : }
788 :
789 0 : HcclResult CcuTransport::GetRmtVarAddrByIndex(uint32_t index, uint64_t& rmtXnAddr) const
790 : {
791 0 : uint32_t rmtXnId{0};
792 0 : CHK_RET(GetRmtXnByIndex(index, rmtXnId));
793 0 : CHK_RET(GetRmtVarAddrByXnId(rmtXnId, rmtXnAddr));
794 0 : HCCL_DEBUG("[CcuTransport][%s] index[%u], xnAddr[%u][0x%llx]", __func__, index, rmtXnId, rmtXnAddr);
795 0 : return HCCL_SUCCESS;
796 : }
797 :
798 0 : HcclResult CcuTransport::GetRmtVarAddrByXnId(const uint32_t rmtXnId, uint64_t& rmtXnAddr) const
799 : {
800 0 : uint64_t xnOffsetCcumAddr = 0;
801 0 : CHK_PRT_RET(
802 : CcuDevMgrImp::GetXnOffsetCcumAddrById(devLogicId_, dieId_, rmtXnId, xnOffsetCcumAddr),
803 : HCCL_ERROR(
804 : "[CcuTransport][%s] Failed to get xn offset address. devLogicId = %d, dieId = %u.", __func__, devLogicId_,
805 : dieId_),
806 : HCCL_E_INTERNAL);
807 0 : const uint64_t rmtResourceAddr = ccuConnection_->GetRmtCcuBufAddr();
808 0 : HCCL_DEBUG(
809 : "[CcuTransport][%s]rmtCcuBufAddr[0x%llx], xnAddr[%u][0x%llx]", __func__, rmtResourceAddr, rmtXnId,
810 : xnOffsetCcumAddr);
811 0 : if (rmtResourceAddr > UINT64_MAX - xnOffsetCcumAddr) {
812 0 : HCCL_ERROR(
813 : "[CcuTransport][%s] failed, CCU resource base address[%llu] is "
814 : "greater than expected, ccu xn offset[%llu], their sum will exceed the range "
815 : "of uint64_t.",
816 : __func__, rmtResourceAddr, xnOffsetCcumAddr);
817 0 : return HCCL_E_INTERNAL;
818 : }
819 0 : rmtXnAddr = rmtResourceAddr + xnOffsetCcumAddr;
820 0 : return HCCL_SUCCESS;
821 : }
822 :
823 8 : HcclResult CcuTransport::GetRmtCcuBufferTokenInfo(uint32_t& rmtTokenId, uint32_t& rmtTokenValue) const
824 : {
825 8 : rmtTokenId = ccuConnection_->GetRmtCcuBufTokenId();
826 8 : rmtTokenValue = ccuConnection_->GetRmtCcuBufTokenValue();
827 8 : return HcclResult::HCCL_SUCCESS;
828 : }
829 :
830 62 : CcuTransport::~CcuTransport() { (void)ReleaseTransRes(); }
831 :
832 1 : std::string CcuTransport::Describe() const
833 : {
834 1 : std::string description = "";
835 :
836 1 : description = Hccl::StringFormat("DieId: %u, ", dieId_);
837 1 : description += transStatus_.Describe();
838 1 : description += Hccl::StringFormat(", LocRes: {%u Ckes, %u Xns}, ", locRes_.ckes.size(), locRes_.xns.size());
839 1 : description += Hccl::StringFormat("RmtRes: {%u Ckes, %u Xns}, ", rmtRes_.ckes.size(), rmtRes_.xns.size());
840 1 : description += Hccl::StringFormat("CkesRes size: %u, ", ckesRes_.size());
841 1 : description += Hccl::StringFormat("XnsRes size: %u.", xnsRes_.size());
842 1 : return description;
843 0 : }
844 :
845 2 : HcclResult CcuTransport::Describe(std::string& dfxMsg)
846 : {
847 2 : CHK_RET(ccuConnection_->Describe(dfxMsg));
848 1 : return HcclResult::HCCL_SUCCESS;
849 : }
850 :
851 1 : void CcuTransport::Clean()
852 : {
853 1 : transStatus_ = TransStatus::INIT;
854 1 : sendData_.clear();
855 : // msg-only(资源不足)transport 无 ccuConnection_，直接跳过 connection 清理
856 1 : if (ccuConnection_ != nullptr) {
857 1 : ccuConnection_->Clean();
858 : }
859 1 : }
860 :
861 0 : HcclResult CcuTransport::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
862 : {
863 0 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
864 : Hccl::RemoteMemCtx<std::unique_ptr<Hccl::RemoteUbRmaBuffer>> remoteMemCtx{
865 0 : cacheValid_, rmtBufferVec_, remoteUserMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
866 0 : CHK_RET(Hccl::GetRemoteUserMems(remoteMemCtx));
867 0 : return HCCL_SUCCESS;
868 0 : }
869 :
870 0 : HcclResult CcuTransport::CheckSocketStatus()
871 : {
872 0 : CHK_PTR_NULL(socket_);
873 0 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
874 0 : auto startTime = std::chrono::steady_clock::now();
875 0 : uint32_t retryCount = 0;
876 : while (true) {
877 : EXCEPTION_HANDLE_BEGIN
878 0 : Hccl::SocketStatus socketStatus = socket_->GetAsyncStatus();
879 0 : if (socketStatus == Hccl::SocketStatus::OK) {
880 : auto elapsed
881 0 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
882 0 : .count();
883 0 : HCCL_INFO("[CcuTransport][%s] success, elapsed[%lld]ms, retryCount[%u]", __func__, elapsed, retryCount);
884 0 : break;
885 : }
886 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout || socketStatus == Hccl::SocketStatus::TIMEOUT) {
887 : auto elapsed
888 0 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
889 0 : .count();
890 0 : HCCL_ERROR(
891 : "[CcuTransport][%s] channel connect timeout after %lld sec, elapsed[%lld]ms, retryCount[%u]", __func__,
892 : timeout.count(), elapsed, retryCount);
893 0 : return HCCL_E_TIMEOUT;
894 : }
895 0 : EXCEPTION_HANDLE_END
896 0 : retryCount++;
897 0 : }
898 0 : return HCCL_SUCCESS;
899 : }
900 :
901 0 : HcclResult CcuTransport::UpdateMemInfo(std::vector<CcuTransport::CclBufferInfo>& bufferVecTemp)
902 : {
903 0 : if (bufferVecTemp.size() == 0) {
904 0 : HCCL_WARNING("[CcuTransport][UpdateMemInfo] bufferNum is 0.");
905 0 : return HCCL_SUCCESS;
906 : }
907 0 : uint32_t totalBufferNum = locBufferInfos_.size() + bufferVecTemp.size();
908 0 : if (UNLIKELY(totalBufferNum > MAX_BUFFER_NUM)) {
909 0 : HCCL_ERROR(
910 : "[CcuTransport][UpdateMemInfo] totalBufferNum[%u] exceeds limit[%u]", totalBufferNum, MAX_BUFFER_NUM);
911 0 : return HCCL_E_PARA;
912 : }
913 0 : HCCL_INFO("[CcuTransport][UpdateMemInfo] bufferNum[%zu]", bufferVecTemp.size());
914 0 : sendData_.clear();
915 0 : Hccl::BinaryStream sendStream;
916 0 : CHK_RET(BufferInfoPack(sendStream, bufferVecTemp));
917 0 : sendStream.Dump(sendData_);
918 0 : u32 sendSize = sendData_.size();
919 : EXCEPTION_HANDLE_BEGIN
920 0 : socket_->SendAsync(&sendSize, sizeof(sendSize));
921 0 : EXCEPTION_HANDLE_END
922 0 : HCCL_INFO(
923 : "[CcuTransport][UpdateMemInfo] Send size[%u] of data success. [%zu] bytes sent.", sendSize, sizeof(sendSize));
924 0 : CHK_RET(CheckSocketStatus());
925 0 : CHK_RET(RecvDataSize());
926 0 : CHK_RET(CheckSocketStatus());
927 0 : CHK_RET(SendConnAndTransInfo());
928 0 : CHK_RET(CheckSocketStatus());
929 0 : CHK_RET(RecvConnAndTransInfo());
930 0 : CHK_RET(CheckSocketStatus());
931 0 : Hccl::BinaryStream recvStream(recvData_);
932 0 : CHK_RET(BufferInfoUnpack(recvStream));
933 0 : locBufferInfos_.insert(locBufferInfos_.end(), bufferVecTemp.begin(), bufferVecTemp.end());
934 : // 流程中已有新增内存数量判断，故执行到此位置一定存在新增内存，需要将标识置位false，使得再次调用GetRemoteMems时重新构造缓存
935 0 : cacheValid_ = false;
936 0 : return HcclResult::HCCL_SUCCESS;
937 0 : }
938 :
939 2 : HcclResult CcuTransport::ResUpdate(std::vector<std::string>& resGroupTags)
940 : {
941 2 : if (resGroupTags.size() == 0) {
942 1 : return HCCL_SUCCESS;
943 : }
944 :
945 3 : for (auto& resGroupTag : resGroupTags) {
946 2 : if (locRes_.cntXns.find(resGroupTag) == locRes_.cntXns.end()) {
947 2 : locRes_.cntXns.insert(std::make_pair(resGroupTag, INVALID_UINT));
948 : }
949 : }
950 :
951 1 : CHK_RET(AppendCntXns());
952 :
953 1 : switch (transStatus_) {
954 0 : case CcuTransport::TransStatus::INIT:
955 0 : break;
956 1 : case CcuTransport::TransStatus::READY:
957 1 : transStatus_ = CcuTransport::TransStatus::SEND_TRANS_RES;
958 1 : break;
959 0 : default:
960 0 : HCCL_ERROR("[CcuTransport][%s] failed, error status[%s].", __func__, transStatus_.Describe().c_str());
961 0 : transStatus_ = CcuTransport::TransStatus::CONNECT_FAILED;
962 0 : break;
963 : }
964 :
965 1 : return HcclResult::HCCL_SUCCESS;
966 : }
967 :
968 : HcclResult
969 4 : CcuTransport::ConstructMsgOnlyTransport(Hccl::Socket* socket, std::unique_ptr<CcuTransport>& impl, CcuResStatus status)
970 : {
971 4 : HCCL_INFO("[CcuTransport][%s] construct msg-only transport, status[%u].", __func__, static_cast<uint8_t>(status));
972 4 : CHK_PTR_NULL(socket);
973 2 : std::vector<CclBufferInfo> emptyBufferInfos{};
974 2 : EXCEPTION_CATCH((impl = std::make_unique<CcuTransport>(socket, nullptr, emptyBufferInfos)), return HCCL_E_PTR);
975 2 : CHK_SMART_PTR_NULL(impl);
976 2 : impl->locResStatus_ = status;
977 2 : impl->transStatus_ = CcuTransport::TransStatus::INIT;
978 2 : return HCCL_SUCCESS;
979 2 : }
980 : } // namespace hcomm
|
