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