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