Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "aicpu_ts_uboe_ub_rtp_channel_helper.h"
12 : #include "endpoint.h"
13 : #include "orion_adpt_utils.h"
14 :
15 : // Orion
16 : #include "virtual_topo.h"
17 : #include "dev_capability.h"
18 : #include "dev_buffer.h"
19 : #include "aicpu_res_package_helper.h"
20 : #include "exchange_ub_buffer_dto.h"
21 : #include "exchange_ub_conn_dto.h"
22 : #include "user_remote_mem_getter.h"
23 : #include "makebufs_helper.h"
24 :
25 : namespace hcomm {
26 :
27 : constexpr u32 SERVER_LISTEN_PORT = 60001;
28 :
29 41 : AicpuTsUboeUbRtpChannelHelper::AicpuTsUboeUbRtpChannelHelper(
30 41 : EndpointHandle endpointHandle, const HcommChannelDesc& channelDesc)
31 41 : : endpointHandle_(endpointHandle),
32 41 : channelDesc_(channelDesc)
33 41 : {}
34 :
35 41 : AicpuTsUboeUbRtpChannelHelper::~AicpuTsUboeUbRtpChannelHelper()
36 : {
37 41 : if (channelDesc_.socket == nullptr && socket_ != nullptr) {
38 0 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
39 0 : socket_ = nullptr;
40 : }
41 41 : }
42 :
43 6 : HcclResult AicpuTsUboeUbRtpChannelHelper::ParseInputParam()
44 : {
45 : // 1. 从 endpointHandle_，获得 localEp_ 和 rdmaHandle_
46 6 : Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
47 6 : CHK_PTR_NULL(localEpPtr);
48 6 : localEp_ = localEpPtr->GetEndpointDesc();
49 6 : rdmaHandle_ = localEpPtr->GetRdmaHandle();
50 :
51 6 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper][%s] localProtocol[%d]", __func__, localEp_.protocol);
52 :
53 : // 2. 从 channelDesc_，获得 remoteEp_, socket_ 和 notifyNum
54 6 : remoteEp_ = channelDesc_.remoteEndpoint;
55 6 : socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
56 6 : notifyNum_ = channelDesc_.notifyNum;
57 6 : commonRes_.bufferVec.clear();
58 :
59 6 : if (channelDesc_.exchangeAllMems) {
60 : // 3. Get memHandles from endpoint
61 2 : HCCL_INFO(
62 : "[AicpuTsUboeUbRtpChannelHelper][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
63 2 : std::shared_ptr<Hccl::LocalUbRmaBuffer>* memHandles = nullptr;
64 2 : uint32_t memHandleNum = 0;
65 2 : CHK_RET(static_cast<HcclResult>(
66 : HcommMemGetAllMemHandles(endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
67 2 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper][%s] Got memHandleNum[%u].", __func__, memHandleNum);
68 4 : for (uint32_t i = 0; i < memHandleNum; ++i) {
69 2 : std::shared_ptr<Hccl::LocalUbRmaBuffer>& localUbRmaBuffer = memHandles[i];
70 2 : CHK_SMART_PTR_NULL(localUbRmaBuffer);
71 2 : Hccl::Buffer* buf = localUbRmaBuffer->GetBuf();
72 2 : CHK_PTR_NULL(buf);
73 2 : HCCL_INFO(
74 : "[AicpuTsUboeUbRtpChannelHelper][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], "
75 : "memType[%d], memInfo[%s].",
76 : __func__, i, static_cast<unsigned long long>(localUbRmaBuffer->GetAddr()),
77 : static_cast<unsigned long long>(localUbRmaBuffer->GetSize()), static_cast<int>(buf->GetMemType()),
78 : buf->GetMemInfo().c_str());
79 2 : commonRes_.bufferVec.push_back(localUbRmaBuffer.get());
80 : }
81 : } else {
82 : // 3. 从 channelDesc 的 memHandle 填充 commonRes_.bufferVec
83 4 : HCCL_INFO(
84 : "[AicpuTsUboeUbRtpChannelHelper][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
85 4 : CHK_RET(MakeRmaBufferVecFromMemHandles(
86 : channelDesc_.memHandles, channelDesc_.memHandleNum, commonRes_.bufferVec, "AicpuTsUboeUbRtpChannelHelper"));
87 : }
88 :
89 6 : return HCCL_SUCCESS;
90 : }
91 :
92 0 : void AicpuTsUboeUbRtpChannelHelper::BuildConn()
93 : {
94 0 : if (BuildConnection() != HCCL_SUCCESS) {
95 0 : HCCL_ERROR("[AicpuTsUboeUbRtpChannelHelper::%s] BuildConnection failed", __func__);
96 : }
97 0 : }
98 :
99 2 : HcclResult AicpuTsUboeUbRtpChannelHelper::BuildNotify()
100 : {
101 2 : localNotifies_.clear();
102 2 : commonRes_.notifyVec.clear();
103 2 : bool devUsed = true;
104 2 : for (uint32_t i = 0; i < notifyNum_; ++i) {
105 0 : std::unique_ptr<Hccl::UbLocalNotify> notifyPtr = nullptr;
106 0 : EXCEPTION_CATCH(notifyPtr = std::make_unique<Hccl::UbLocalNotify>(rdmaHandle_, devUsed), return HCCL_E_PTR);
107 0 : commonRes_.notifyVec.push_back(notifyPtr.get());
108 0 : localNotifies_.push_back(std::move(notifyPtr));
109 0 : }
110 2 : return HCCL_SUCCESS;
111 : }
112 :
113 2 : HcclResult AicpuTsUboeUbRtpChannelHelper::BuildDrainResource()
114 : {
115 : // 申请创建channel drain阻塞等待的相关资源
116 : // 申请notify作为read的落点
117 2 : bool devUsed = true;
118 2 : EXCEPTION_CATCH(drainNotify_ = std::make_unique<Hccl::UbLocalNotify>(rdmaHandle_, devUsed), return HCCL_E_PTR);
119 2 : HCCL_INFO(
120 : "[AicpuTsUboeUbRtpChannelHelper][%s] drainNotify created: %s", __func__, drainNotify_->Describe().c_str());
121 :
122 : // 常量1内存供远端读取
123 2 : u32 notifySize = Hccl::DevCapability::GetInstance().GetNotifySize();
124 :
125 2 : std::shared_ptr<Hccl::DevBuffer> constMem;
126 2 : EXCEPTION_CATCH(constMem = std::make_shared<Hccl::DevBuffer>(notifySize), return HCCL_E_PTR);
127 :
128 2 : Hccl::HrtMemcpy(
129 4 : reinterpret_cast<void*>(constMem->GetAddr()), constMem->GetSize(), &NORMAL_NOTIFY_VAL,
130 : sizeof(NORMAL_NOTIFY_VAL), Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
131 :
132 2 : EXCEPTION_CATCH(drainBuffer_ = std::make_unique<Hccl::LocalUbRmaBuffer>(constMem, rdmaHandle_), return HCCL_E_PTR);
133 2 : HCCL_INFO(
134 : "[AicpuTsUboeUbRtpChannelHelper][%s] drain buffer created: addr[0x%llx], size[%zu]", __func__,
135 : static_cast<unsigned long long>(drainBuffer_->GetAddr()), drainBuffer_->GetSize());
136 :
137 2 : return HCCL_SUCCESS;
138 2 : }
139 :
140 2 : HcclResult AicpuTsUboeUbRtpChannelHelper::BuildSocket()
141 : {
142 2 : if (socket_ != nullptr) {
143 2 : return HCCL_SUCCESS;
144 : }
145 0 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper][%s] socket ptr is NULL, rebuildSocket", __func__);
146 :
147 0 : Hccl::IpAddress ipaddr{};
148 0 : CHK_RET(CommAddrToIpAddress(localEp_.commAddr, ipaddr));
149 0 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
150 0 : Hccl::PortData localPort = Hccl::PortData(static_cast<Hccl::RankId>(localEp_.loc.device.devPhyId), type, 0, ipaddr);
151 : Hccl::SocketHandle socketHandle
152 0 : = Hccl::SocketHandleManager::GetInstance().Create(localEp_.loc.device.devPhyId, localPort);
153 0 : EXCEPTION_CATCH(
154 : serverSocket_ = std::make_unique<Hccl::Socket>(
155 : socketHandle, ipaddr, SERVER_LISTEN_PORT, ipaddr, "server", Hccl::SocketRole::SERVER,
156 : Hccl::NicType::DEVICE_NIC_TYPE),
157 : return HCCL_E_PARA);
158 0 : HCCL_INFO(
159 : "[AicpuTsUboeUbRtpChannelHelper][%s] listen_socket_info[%s]", __func__, serverSocket_->Describe().c_str());
160 0 : EXCEPTION_CATCH(serverSocket_->Listen(), return HCCL_E_INTERNAL);
161 :
162 0 : Hccl::LinkData linkData = BuildDefaultLinkData();
163 0 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
164 0 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper][%s] built linkData: %s", __func__, linkData.Describe().c_str());
165 : std::string socketTag
166 0 : = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
167 0 : bool noRankId = true;
168 0 : Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, socketTag, noRankId);
169 0 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
170 0 : isRecvFirst_ = socket_->GetRole() == Hccl::SocketRole::CLIENT ? true : false;
171 :
172 0 : return HCCL_SUCCESS;
173 0 : }
174 :
175 2 : HcclResult AicpuTsUboeUbRtpChannelHelper::GetNotifyNum(uint32_t* notifyNum) const
176 : {
177 2 : *notifyNum = this->notifyNum_;
178 2 : return HCCL_SUCCESS;
179 : }
180 :
181 7 : HcclResult AicpuTsUboeUbRtpChannelHelper::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
182 : {
183 7 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
184 : Hccl::RemoteMemCtx<std::unique_ptr<Hccl::RemoteUbRmaBuffer>> remoteMemCtx{
185 7 : cacheValid_, rmtBufferVec_, remoteUserMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
186 7 : CHK_RET(GetRemoteUserMems(remoteMemCtx));
187 4 : return HCCL_SUCCESS;
188 7 : }
189 :
190 7 : bool AicpuTsUboeUbRtpChannelHelper::IsSocketReady()
191 : {
192 7 : if (socket_ == nullptr) {
193 1 : HCCL_ERROR("[%s] socket is nullptr, please check", __func__);
194 1 : channelStatus = ChannelStatus::INVALID;
195 1 : return false;
196 : }
197 :
198 6 : Hccl::SocketStatus socketStatus = socket_->GetAsyncStatus();
199 6 : if (socketStatus == Hccl::SocketStatus::OK) {
200 6 : channelStatus = ChannelStatus::SOCKET_OK;
201 6 : return true;
202 0 : } else if (socketStatus == Hccl::SocketStatus::TIMEOUT) {
203 0 : channelStatus = ChannelStatus::SOCKET_TIMEOUT;
204 0 : return false;
205 : }
206 :
207 0 : return false;
208 : }
209 :
210 2 : bool AicpuTsUboeUbRtpChannelHelper::IsResReady()
211 : {
212 2 : for (auto& it : commonRes_.connVec) {
213 0 : if (it == nullptr) {
214 0 : Hccl::THROW<Hccl::InternalException>(
215 : "[AicpuTsUboeUbRtpChannelHelper::%s] failed, connection pointer is nullptr", __func__);
216 : }
217 0 : Hccl::RmaConnType connType = it->GetRmaConnType();
218 0 : if (connType != Hccl::RmaConnType::UB) {
219 0 : Hccl::THROW<Hccl::InternalException>(
220 : "[AicpuTsUboeUbRtpChannelHelper::%s] connection type[%s] is not ub", __func__,
221 0 : connType.Describe().c_str());
222 : }
223 :
224 0 : auto status = it->GetStatus();
225 0 : if (status != Hccl::RmaConnStatus::EXCHANGEABLE && status != Hccl::RmaConnStatus::READY) {
226 0 : return false;
227 : }
228 : }
229 2 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] all resources ready.", __func__);
230 2 : return true;
231 : }
232 :
233 1 : bool AicpuTsUboeUbRtpChannelHelper::IsConnsReady()
234 : {
235 1 : for (u32 i = 0; i < connNum_; i++) {
236 0 : if (commonRes_.connVec[i]->GetStatus() != Hccl::RmaConnStatus::READY) {
237 0 : return false;
238 : }
239 : }
240 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] conns are ready.", __func__);
241 1 : return true;
242 : }
243 :
244 3 : void AicpuTsUboeUbRtpChannelHelper::NotifyVecPack(Hccl::BinaryStream& binaryStream)
245 : {
246 3 : binaryStream << notifyNum_;
247 3 : u32 pos = 0;
248 3 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] pack notify size[%d]", __func__, commonRes_.notifyVec.size());
249 3 : for (auto& it : commonRes_.notifyVec) {
250 0 : binaryStream << pos;
251 0 : std::unique_ptr<Hccl::Serializable> dto = it->GetExchangeDto();
252 0 : dto->Serialize(binaryStream);
253 0 : HCCL_INFO(
254 : "[AicpuTsUboeUbRtpChannelHelper::%s] pack notify pos=%u, dto %s", __func__, pos, dto->Describe().c_str());
255 0 : pos++;
256 0 : }
257 3 : }
258 :
259 5 : void AicpuTsUboeUbRtpChannelHelper::BufferVecPack(
260 : Hccl::BinaryStream& binaryStream, std::vector<Hccl::LocalRmaBuffer*>& bufferVec)
261 : {
262 5 : binaryStream << static_cast<u32>(bufferVec.size());
263 5 : u32 pos = 0;
264 5 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] pack buffer size[%d]", __func__, bufferVec.size());
265 7 : for (auto& it : bufferVec) {
266 2 : binaryStream << pos;
267 2 : if (it != nullptr) {
268 2 : std::unique_ptr<Hccl::Serializable> dto = it->GetExchangeDto();
269 2 : dto->Serialize(binaryStream);
270 2 : HCCL_INFO(
271 : "[AicpuTsUboeUbRtpChannelHelper::%s] pack buffer pos=%u dto %s", __func__, pos,
272 : dto->Describe().c_str());
273 2 : } else {
274 0 : Hccl::ExchangeUbBufferDto exchangeDto;
275 0 : exchangeDto.Serialize(binaryStream);
276 0 : HCCL_INFO(
277 : "[AicpuTsUboeUbRtpChannelHelper::%s] pack buffer pos=%u, dto is null %s", __func__, pos,
278 : exchangeDto.Describe().c_str());
279 0 : }
280 2 : pos++;
281 : }
282 5 : }
283 :
284 1 : void AicpuTsUboeUbRtpChannelHelper::DrainBufferPack(Hccl::BinaryStream& binaryStream)
285 : {
286 : // 只需交换常量buffer信息供对端读
287 1 : HCCL_INFO("start pack drain buffer");
288 1 : if (drainBuffer_ != nullptr) { // 非空的buffer，从buffer中获取 dto
289 0 : std::unique_ptr<Hccl::Serializable> dto = drainBuffer_->GetExchangeDto();
290 0 : dto->Serialize(binaryStream);
291 0 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] pack drain buffer dto %s", __func__, dto->Describe().c_str());
292 0 : } else { // 空的buffer，dto所有字段为0(size=0)
293 1 : Hccl::ExchangeUbBufferDto exchangeDto;
294 1 : exchangeDto.Serialize(binaryStream);
295 1 : HCCL_INFO(
296 : "[AicpuTsUboeUbRtpChannelHelper::%s] pack drain buffer dto is null %s", __func__,
297 : exchangeDto.Describe().c_str());
298 1 : }
299 1 : }
300 :
301 3 : void AicpuTsUboeUbRtpChannelHelper::ConnVecPack(Hccl::BinaryStream& binaryStream)
302 : {
303 3 : binaryStream << connNum_;
304 3 : u32 pos = 0;
305 3 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] pack conn size[%d]", __func__, commonRes_.connVec.size());
306 3 : for (auto& it : commonRes_.connVec) {
307 0 : binaryStream << pos;
308 0 : std::unique_ptr<Hccl::Serializable> dto = it->GetExchangeDto();
309 0 : dto->Serialize(binaryStream);
310 0 : HCCL_INFO(
311 : "[AicpuTsUboeUbRtpChannelHelper::%s] pack connection pos=%u, dto %s", __func__, pos,
312 : dto->Describe().c_str());
313 0 : pos++;
314 0 : }
315 3 : }
316 :
317 1 : void AicpuTsUboeUbRtpChannelHelper::SendDataSize()
318 : {
319 1 : sendData_.clear();
320 1 : bufferNum_ = commonRes_.bufferVec.size();
321 1 : connNum_ = commonRes_.connVec.size();
322 :
323 1 : HCCL_INFO("notifyNum=%u, bufferNum=%u, connNum=%u", notifyNum_, bufferNum_, connNum_);
324 :
325 1 : Hccl::BinaryStream binaryStream;
326 1 : NotifyVecPack(binaryStream);
327 1 : BufferVecPack(binaryStream, commonRes_.bufferVec);
328 1 : DrainBufferPack(binaryStream);
329 1 : ConnVecPack(binaryStream);
330 :
331 1 : binaryStream.Dump(sendData_);
332 1 : u32 sendSize = sendData_.size();
333 :
334 : // 发送数据包尺寸
335 1 : socket_->SendAsync(&sendSize, sizeof(sendSize));
336 1 : HCCL_INFO(
337 : "[AicpuTsUboeUbRtpChannelHelper::%s] Send size[%u] of data success. [%zu] bytes sent.", __func__, sendSize,
338 : sizeof(sendSize));
339 1 : }
340 :
341 1 : void AicpuTsUboeUbRtpChannelHelper::RecvDataSize()
342 : {
343 : // 接收数据包尺寸
344 1 : socket_->RecvAsync(reinterpret_cast<u8*>(&recvDataSize_), sizeof(recvDataSize_));
345 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] Receive Data Size", __func__);
346 1 : }
347 :
348 1 : void AicpuTsUboeUbRtpChannelHelper::SendExchangeData()
349 : {
350 1 : socket_->SendAsync(sendData_.data(), sendData_.size());
351 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] send data, size=%llu", __func__, sendData_.size());
352 1 : }
353 :
354 1 : void AicpuTsUboeUbRtpChannelHelper::RecvExchangeData()
355 : {
356 1 : recvData_.resize(recvDataSize_);
357 1 : socket_->RecvAsync(reinterpret_cast<u8*>(recvData_.data()), recvData_.size());
358 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] recv data", __func__);
359 1 : }
360 :
361 1 : bool AicpuTsUboeUbRtpChannelHelper::RecvDataProcess()
362 : {
363 1 : HCCL_INFO("RecvDataProcess: size=%llu, recvDataSize=%u", recvData_.size(), recvDataSize_);
364 1 : Hccl::BinaryStream binaryStream(recvData_);
365 1 : RmtBufferVecUnpackProc(notifyNum_, binaryStream, rmtNotifyVec_, UboeRmtBufType::NOTIFY);
366 1 : RmtBufferVecUnpackProc(bufferNum_, binaryStream, rmtBufferVec_, UboeRmtBufType::BUFFER);
367 1 : RmtDrainBufferUnpackProc(binaryStream);
368 2 : return ConnVecUnpackProc(binaryStream);
369 1 : }
370 :
371 3 : void AicpuTsUboeUbRtpChannelHelper::RmtBufferVecUnpackProc(
372 : u32 locNum, Hccl::BinaryStream& binaryStream, RemoteBufferVec& bufferVec, UboeRmtBufType type)
373 : {
374 : u32 rmtNum;
375 3 : binaryStream >> rmtNum;
376 3 : if (type == UboeRmtBufType::BUFFER && rmtNum > MAX_BUFFER_NUM) {
377 0 : MACRO_THROW(
378 : Hccl::InvalidParamsException,
379 : Hccl::StringFormat(
380 : "[AicpuTsUboeUbRtpChannelHelper][RmtBufferVecUnpackProc] rmtNum[%u] exceeds limit[%u]", rmtNum,
381 : MAX_BUFFER_NUM));
382 : }
383 :
384 3 : HCCL_INFO("unpack %s, locNum=%u, rmtNum=%u", type.Describe().c_str(), locNum, rmtNum);
385 :
386 4 : for (u32 i = 0; i < rmtNum; i++) {
387 : u32 pos;
388 1 : binaryStream >> pos;
389 1 : Hccl::ExchangeUbBufferDto dto;
390 1 : dto.Deserialize(binaryStream);
391 1 : if (bufferVec.size() > pos) {
392 0 : continue;
393 : }
394 1 : HCCL_INFO("unpack %s pos=%u, dto %s", type.Describe().c_str(), pos, dto.Describe().c_str());
395 1 : if (dto.size == 0) {
396 0 : HCCL_INFO("unpack nullptr, pos=%u", pos);
397 0 : bufferVec.push_back(nullptr);
398 : } else {
399 1 : bufferVec.push_back(std::make_unique<Hccl::RemoteUbRmaBuffer>(rdmaHandle_, dto));
400 1 : HCCL_INFO("unpack buffer pos=%u, rmtRmaBuffer=%s", pos, bufferVec.back()->Describe().c_str());
401 : }
402 1 : }
403 3 : }
404 :
405 1 : void AicpuTsUboeUbRtpChannelHelper::RmtDrainBufferUnpackProc(Hccl::BinaryStream& binaryStream)
406 : {
407 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] start unpack drain buffer", __func__);
408 1 : Hccl::ExchangeUbBufferDto dto;
409 1 : dto.Deserialize(binaryStream);
410 :
411 1 : if (dto.size == 0) {
412 1 : rmtDrainBuffer_ = nullptr;
413 1 : HCCL_WARNING("[AicpuTsUboeUbRtpChannelHelper::%s] unpack drain buffer dto is null", __func__);
414 : } else {
415 0 : rmtDrainBuffer_ = std::make_unique<Hccl::RemoteUbRmaBuffer>(rdmaHandle_, dto);
416 0 : HCCL_INFO(
417 : "[AicpuTsUboeUbRtpChannelHelper::%s] unpack drain buffer rmtDrainBuffer=%s", __func__,
418 : rmtDrainBuffer_->Describe().c_str());
419 : }
420 1 : }
421 :
422 1 : bool AicpuTsUboeUbRtpChannelHelper::ConnVecUnpackProc(Hccl::BinaryStream& binaryStream)
423 : {
424 : u32 rmtConnNum;
425 1 : binaryStream >> rmtConnNum;
426 1 : HCCL_INFO("start unpack conn connNum=%u, rmtConnNum=%u", connNum_, rmtConnNum);
427 1 : if (connNum_ != rmtConnNum) {
428 0 : MACRO_THROW(
429 : Hccl::InvalidParamsException,
430 : Hccl::StringFormat("connNum=%u is not equal to rmtConnNum=%u", connNum_, rmtConnNum));
431 : }
432 :
433 1 : bool result = false;
434 1 : for (u32 i = 0; i < rmtConnNum; i++) {
435 : u32 pos;
436 0 : binaryStream >> pos;
437 0 : Hccl::ExchangeUbConnDto rmtDto;
438 0 : rmtDto.Deserialize(binaryStream);
439 0 : HCCL_INFO("unpack connection pos=%u dto %s", pos, rmtDto.Describe().c_str());
440 0 : if (commonRes_.connVec[i]->GetStatus() != Hccl::RmaConnStatus::READY) {
441 0 : HCCL_INFO(
442 : "parse and import pos=%u, rmt dto to connection[%s]", pos, commonRes_.connVec[i]->Describe().c_str());
443 0 : commonRes_.connVec[i]->ParseRmtExchangeDto(rmtDto);
444 0 : commonRes_.connVec[i]->ImportRmtDto();
445 0 : result = true;
446 : }
447 0 : }
448 1 : return result;
449 : }
450 :
451 2 : static HcclResult SetUboeModuleDataName(Hccl::ModuleData& module, const std::string& name)
452 : {
453 2 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
454 2 : if (ret != 0) {
455 0 : HCCL_ERROR("[SetModuleDataName] strcpy_s name %s failed", name.c_str());
456 0 : return HCCL_E_INTERNAL;
457 : }
458 2 : return HCCL_SUCCESS;
459 : }
460 :
461 2 : std::vector<char> AicpuTsUboeUbRtpChannelHelper::GetNotifyUniqueIds()
462 : {
463 2 : HCCL_INFO("start packing all notify uniqueIds");
464 2 : std::vector<char> result(0);
465 4 : for (auto& it : commonRes_.notifyVec) {
466 2 : HCCL_INFO("AicpuTsUboeUbRtpChannelHelper Notify %s", it->Describe().c_str());
467 2 : auto uniqueId = it->GetUniqueId();
468 2 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
469 2 : }
470 2 : return result;
471 0 : }
472 :
473 8 : std::vector<char> AicpuTsUboeUbRtpChannelHelper::GetSingleRmtBufferUniqueId(
474 : u64 addr, u64 size, u32 tokenId, u32 tokenValue, u32 notifyId) const
475 : {
476 8 : Hccl::BinaryStream binaryStream;
477 8 : binaryStream << addr;
478 8 : binaryStream << size;
479 8 : binaryStream << tokenId;
480 8 : binaryStream << tokenValue;
481 8 : binaryStream << notifyId;
482 8 : HCCL_INFO("AicpuTsUboeUbRtpChannelHelper RmtBuffer[addr=0x%llx, size=0x%llx, notifyId=%u]", addr, size, notifyId);
483 8 : std::vector<char> result;
484 8 : binaryStream.Dump(result);
485 8 : return result;
486 8 : }
487 :
488 : std::vector<char>
489 4 : AicpuTsUboeUbRtpChannelHelper::GetSingleLocBufferUniqueId(u64 addr, u64 size, u32 tokenId, u32 tokenValue) const
490 : {
491 4 : Hccl::BinaryStream binaryStream;
492 4 : binaryStream << addr;
493 4 : binaryStream << size;
494 4 : binaryStream << tokenId;
495 4 : binaryStream << tokenValue;
496 4 : HCCL_INFO("AicpuTsUboeUbRtpChannelHelper LocBuffer[addr=0x%llx, size=0x%llx]", addr, size);
497 4 : std::vector<char> result;
498 4 : binaryStream.Dump(result);
499 4 : return result;
500 4 : }
501 :
502 : std::vector<char>
503 4 : AicpuTsUboeUbRtpChannelHelper::GetRmtBufferUniqueIds(RemoteBufferVec& bufferVec, UboeRmtBufType type) const
504 : {
505 4 : HCCL_INFO("start packing all remote buffer %s uniqueIds", type.Describe().c_str());
506 4 : std::vector<char> result(0);
507 8 : for (auto& it : bufferVec) {
508 4 : std::vector<char> uniqueId;
509 4 : if (it != nullptr) {
510 16 : uniqueId = GetSingleRmtBufferUniqueId(
511 12 : it->GetAddr(), it->GetSize(), it->GetTokenId(), it->GetTokenValue(), it->GetNotifyId());
512 4 : HCCL_INFO("AicpuTsUboeUbRtpChannelHelper::GetRmtBufferUniqueIds, %s", it->Describe().c_str());
513 : } else {
514 0 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX);
515 0 : HCCL_INFO("AicpuTsUboeUbRtpChannelHelper::GetRmtBufferUniqueIds, null buffer");
516 : }
517 4 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
518 4 : }
519 4 : return result;
520 0 : }
521 :
522 : std::vector<char>
523 2 : AicpuTsUboeUbRtpChannelHelper::GetLocBufferUniqueIds(LocalBufferVec& bufferVec, UboeRmtBufType type) const
524 : {
525 2 : HCCL_INFO("start packing all local buffer %s uniqueIds", type.Describe().c_str());
526 2 : std::vector<char> result(0);
527 6 : for (auto& it : bufferVec) {
528 4 : std::vector<char> uniqueId;
529 4 : if (it != nullptr) {
530 2 : uniqueId = GetSingleLocBufferUniqueId(it->GetAddr(), it->GetSize(), it->GetTokenId(), it->GetTokenValue());
531 2 : HCCL_INFO("UbMemTransport::GetLocBufferUniqueIds, %s", it->Describe().c_str());
532 : } else {
533 2 : uniqueId = GetSingleLocBufferUniqueId(0, 0, 0, 0);
534 2 : HCCL_INFO("UbMemTransport::GetLocBufferUniqueIds, null buffer");
535 : }
536 4 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
537 4 : }
538 2 : return result;
539 0 : }
540 :
541 2 : std::vector<char> AicpuTsUboeUbRtpChannelHelper::GetDrainUniqueIds() const
542 : {
543 2 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] start get drain resource uniqueIds", __func__);
544 2 : std::vector<char> result(0);
545 2 : std::vector<char> uniqueId;
546 :
547 : // drain notify UniqueId
548 2 : if (drainNotify_ != nullptr) {
549 0 : auto dto = drainNotify_->GetExchangeDto();
550 0 : Hccl::ExchangeUbBufferDto* rawDto = static_cast<Hccl::ExchangeUbBufferDto*>(dto.get());
551 0 : uniqueId = GetSingleRmtBufferUniqueId(
552 0 : rawDto->addr, rawDto->size, rawDto->tokenId, rawDto->tokenValue, rawDto->notifyId);
553 0 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] %s", __func__, drainNotify_->Describe().c_str());
554 0 : } else {
555 2 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
556 2 : HCCL_WARNING("[AicpuTsUboeUbRtpChannelHelper::%s] drainNotify_ null buffer", __func__);
557 : }
558 2 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
559 :
560 : // drain buffer UniqueId
561 2 : if (rmtDrainBuffer_ != nullptr) {
562 4 : uniqueId = GetSingleRmtBufferUniqueId(
563 2 : rmtDrainBuffer_->GetAddr(), rmtDrainBuffer_->GetSize(), rmtDrainBuffer_->GetTokenId(),
564 1 : rmtDrainBuffer_->GetTokenValue(), rmtDrainBuffer_->GetNotifyId());
565 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] %s", __func__, rmtDrainBuffer_->Describe().c_str());
566 : } else {
567 1 : uniqueId = GetSingleRmtBufferUniqueId(0, 0, 0, 0, UINT32_MAX); // 填充一个空的buffer
568 1 : HCCL_WARNING("[AicpuTsUboeUbRtpChannelHelper::%s] rmtDrainBuffer_ null buffer", __func__);
569 : }
570 2 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
571 :
572 2 : return result;
573 2 : }
574 :
575 2 : std::vector<char> AicpuTsUboeUbRtpChannelHelper::GetConnUniqueIds()
576 : {
577 2 : HCCL_INFO("start packing all conn uniqueIds");
578 2 : std::vector<char> result(0);
579 4 : for (auto& it : commonRes_.connVec) {
580 2 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] conn[%s]", __func__, it->Describe().c_str());
581 2 : auto uniqueId = it->GetUniqueId();
582 2 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
583 2 : }
584 2 : return result;
585 0 : }
586 :
587 3 : std::vector<char> AicpuTsUboeUbRtpChannelHelper::GetUniqueIdV2()
588 : {
589 3 : if (channelStatus != ChannelStatus::READY) {
590 1 : MACRO_THROW(
591 : Hccl::InternalException,
592 : Hccl::StringFormat(
593 : "channel status[%d] is not ready[%d], please check.", channelStatus, ChannelStatus::READY));
594 : }
595 2 : u32 type = static_cast<u32>(Hccl::TransportType::UB);
596 2 : Hccl::BinaryStream binaryStream;
597 2 : binaryStream << type;
598 2 : binaryStream << notifyNum_;
599 2 : binaryStream << bufferNum_;
600 2 : binaryStream << static_cast<u32>(rmtBufferVec_.size());
601 2 : binaryStream << connNum_;
602 :
603 2 : auto notifyUniqueIds = GetNotifyUniqueIds();
604 2 : binaryStream << notifyUniqueIds;
605 :
606 2 : auto rmtNotifyUniqueIds = GetRmtBufferUniqueIds(rmtNotifyVec_, UboeRmtBufType::NOTIFY);
607 2 : binaryStream << rmtNotifyUniqueIds;
608 :
609 6 : for (auto& it : commonRes_.bufferVec) {
610 4 : locBufferVec_.emplace_back(reinterpret_cast<Hccl::LocalUbRmaBuffer*>(it));
611 : }
612 :
613 2 : auto locBufferUniqueIds = GetLocBufferUniqueIds(locBufferVec_, UboeRmtBufType::BUFFER);
614 2 : binaryStream << locBufferUniqueIds;
615 :
616 2 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds(rmtBufferVec_, UboeRmtBufType::BUFFER);
617 2 : binaryStream << rmtBufferUniqueIds;
618 :
619 2 : auto drainUniqueIds = GetDrainUniqueIds();
620 2 : binaryStream << drainUniqueIds;
621 :
622 2 : auto connUniqueIds = GetConnUniqueIds();
623 2 : binaryStream << connUniqueIds;
624 :
625 2 : std::vector<char> result;
626 2 : binaryStream.Dump(result);
627 2 : return result;
628 2 : }
629 :
630 2 : HcclResult AicpuTsUboeUbRtpChannelHelper::H2DResPack(std::vector<char>& buffer)
631 : {
632 2 : std::vector<Hccl::ModuleData> dataVec;
633 2 : dataVec.resize(Hccl::AicpuResMgrType::__COUNT__);
634 :
635 2 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
636 4 : CHK_RET(SetUboeModuleDataName(dataVec[resType], "AicpuTsUboeUbRtpChannelHelper"));
637 :
638 2 : std::vector<char> result;
639 2 : Hccl::BinaryStream binaryStream;
640 2 : binaryStream << GetUniqueIdV2();
641 :
642 2 : binaryStream.Dump(result);
643 :
644 2 : dataVec[resType].data = result;
645 :
646 : Hccl::AicpuResPackageHelper helper;
647 2 : buffer = helper.GetPackedData(dataVec);
648 2 : HCCL_INFO(
649 : "[AicpuTsUboeUbRtpChannelHelper][%s] Pack Buffer data[%p], Pack Buffer size[%zu].", __func__, buffer.data(),
650 : buffer.size());
651 2 : return HCCL_SUCCESS;
652 2 : }
653 :
654 0 : HcclResult AicpuTsUboeUbRtpChannelHelper::Clean()
655 : {
656 : // 该模式当前不支持N秒快恢
657 0 : return HCCL_SUCCESS;
658 : }
659 :
660 0 : HcclResult AicpuTsUboeUbRtpChannelHelper::Resume()
661 : {
662 : // 该模式当前不支持N秒快恢
663 0 : return HCCL_SUCCESS;
664 : }
665 :
666 1 : HcclResult AicpuTsUboeUbRtpChannelHelper::NotifyRecord([[maybe_unused]] const uint32_t remoteNotifyIdx)
667 : {
668 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] not supported yet.", __func__);
669 1 : return HCCL_E_NOT_SUPPORT;
670 : }
671 :
672 1 : HcclResult AicpuTsUboeUbRtpChannelHelper::NotifyWait(
673 : [[maybe_unused]] const uint32_t localNotifyIdx, [[maybe_unused]] const uint32_t timeout)
674 : {
675 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] not supported yet.", __func__);
676 1 : return HCCL_E_NOT_SUPPORT;
677 : }
678 :
679 1 : HcclResult AicpuTsUboeUbRtpChannelHelper::WriteWithNotify(
680 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] const uint64_t len,
681 : [[maybe_unused]] uint32_t remoteNotifyIdx)
682 : {
683 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] not supported yet.", __func__);
684 1 : return HCCL_E_NOT_SUPPORT;
685 : }
686 :
687 1 : HcclResult AicpuTsUboeUbRtpChannelHelper::Write(
688 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
689 : {
690 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] not supported yet.", __func__);
691 1 : return HCCL_E_NOT_SUPPORT;
692 : }
693 :
694 1 : HcclResult AicpuTsUboeUbRtpChannelHelper::Read(
695 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
696 : {
697 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] not supported yet.", __func__);
698 1 : return HCCL_E_NOT_SUPPORT;
699 : }
700 :
701 1 : HcclResult AicpuTsUboeUbRtpChannelHelper::ChannelFence()
702 : {
703 1 : HCCL_INFO("[AicpuTsUboeUbRtpChannelHelper::%s] not supported yet.", __func__);
704 1 : return HCCL_E_NOT_SUPPORT;
705 : }
706 :
707 : } // namespace hcomm
|
