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