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_channel.h"
12 : #include "orion_adpt_utils.h"
13 : #include "hcomm_res_mgr.h"
14 : #include "env_config/env_config_v2.h"
15 : #include "makebufs_helper.h"
16 : #include "endpoint.h"
17 :
18 : // Orion
19 : #include "adapter_rts_common.h"
20 : #include "topo_common_types.h"
21 : #include "rdma_handle_manager.h"
22 :
23 : namespace hcomm {
24 :
25 25 : AicpuTsUboeChannel::AicpuTsUboeChannel(EndpointHandle endpointHandle, const HcommChannelDesc& channelDesc)
26 25 : : AicpuTsUboeUbRtpChannelHelper(endpointHandle, channelDesc)
27 : {
28 25 : type_ = static_cast<u32>(Hccl::TransportType::UBoE);
29 25 : }
30 :
31 25 : AicpuTsUboeChannel::~AicpuTsUboeChannel() = default;
32 :
33 2 : HcclResult AicpuTsUboeChannel::Init()
34 : {
35 : s32 devLogicId;
36 2 : CHK_RET(hrtGetDevice(&devLogicId));
37 2 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
38 2 : CHK_RET(ParseInputParam());
39 2 : CHK_RET(BuildSocket());
40 2 : CHK_RET(BuildNotify());
41 2 : CHK_RET(BuildDrainResource());
42 : /*
43 : HccpRaGetDevBaseAttr
44 : 获取urma read/write 单个wr的最大传输数据大小
45 : 调用前,rdmaHandle_要在ParseInputParam中被赋值好,之后BuildConnection会使用获取的属性
46 : uboe的BuildConnection不再Init里面执行，Init之后会有单独流程建链
47 : */
48 2 : CHK_RET(HccpRaGetDevBaseAttr(rdmaHandle_, &devBaseAttr_));
49 :
50 2 : return HCCL_SUCCESS;
51 : }
52 :
53 0 : HcclResult AicpuTsUboeChannel::BuildConnection()
54 : {
55 0 : UbConnBuildContext ctx;
56 0 : CHK_RET(PrepareUbConnBuildContext(localEp_, remoteEp_, channelDesc_, ctx));
57 :
58 0 : Hccl::OpMode opMode = Hccl::OpMode::OPBASE;
59 0 : bool devUsed = true; // aicpu 为 true
60 0 : HCCL_INFO(
61 : "[AicpuTsUboeChannel::%s] LinkProtocol[%s], locIpv4Addr[%s], rmtIpv4Addr[%s]", __func__,
62 : ctx.protocol.Describe().c_str(), ctx.locAddr.Describe().c_str(), ctx.rmtAddr.Describe().c_str());
63 0 : HCCL_INFO(
64 : "[AicpuTsUboeChannel::%s] locAddr_[%s], rmtAddr_[%s]", __func__, locAddr_.Describe().c_str(),
65 : rmtAddr_.Describe().c_str());
66 :
67 : // UBOE 协议对应 HCOMM_TA_RTP_UBOE_TIMEOUT，由 base_comm 从环境变量获取后传入
68 0 : u8 taTimeOut = 0;
69 0 : uint32_t taTimeOutValue = 0;
70 0 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaRtpUboeTimeOut(taTimeOutValue));
71 0 : taTimeOut = static_cast<u8>(taTimeOutValue);
72 0 : std::unique_ptr<Hccl::DevUbConnection> ubConn = std::make_unique<Hccl::DevUbUboeConnection>(
73 0 : rdmaHandle_, locAddr_, rmtAddr_, opMode, devUsed, Hccl::HrtUbJfcMode::STARS_POLL, ctx.locAddr, ctx.rmtAddr,
74 0 : ctx.qosPre, taTimeOut);
75 0 : CHK_SMART_PTR_NULL(ubConn);
76 :
77 0 : if (devBaseAttr_.maxReadSize == 0 || devBaseAttr_.maxWriteSize == 0) {
78 0 : HCCL_ERROR(
79 : "[AicpuTsUboeChannel][%s] maxReadSize[%u] or maxWriteSize[%u] must not be zero", __func__,
80 : devBaseAttr_.maxReadSize, devBaseAttr_.maxWriteSize);
81 0 : return HCCL_E_PARA;
82 : }
83 0 : ubConn->SetMaxReadSize(devBaseAttr_.maxReadSize);
84 0 : ubConn->SetMaxWriteSize(devBaseAttr_.maxWriteSize);
85 0 : HCCL_INFO(
86 : "[AicpuTsUboeChannel][%s] maxReadSize[%u], maxWriteSize[%u]", __func__, devBaseAttr_.maxReadSize,
87 : devBaseAttr_.maxWriteSize);
88 :
89 0 : commonRes_.connVec.clear();
90 0 : commonRes_.connVec.emplace_back(ubConn.get());
91 0 : connections_.clear();
92 0 : connections_.push_back(std::move(ubConn));
93 0 : return HCCL_SUCCESS;
94 0 : }
95 :
96 0 : void AicpuTsUboeChannel::EidPack()
97 : {
98 0 : Hccl::IpAddress locIpv4Addr;
99 0 : CommAddrToIpAddress(localEp_.commAddr, locIpv4Addr);
100 0 : Hccl::RdmaHandleManager::GetInstance().GetEidByIpv4Addr(locIpv4Addr, locAddr_);
101 0 : sendEidData_ = locAddr_.GetUniqueId();
102 0 : HCCL_INFO(
103 : "[AicpuTsUboeChannel::%s] locIpv4Addr[%s], locAddr_[%s], sendEidData_ size[%u]", __func__,
104 : locIpv4Addr.Describe().c_str(), locAddr_.Describe().c_str(), sendEidData_.size());
105 0 : }
106 :
107 0 : void AicpuTsUboeChannel::SendEidData()
108 : {
109 0 : EidPack();
110 0 : socket_->SendAsync(sendEidData_.data(), sendEidData_.size());
111 0 : HCCL_INFO("[AicpuTsUboeChannel::%s] send eid data, size=%llu", __func__, sendEidData_.size());
112 0 : }
113 :
114 0 : void AicpuTsUboeChannel::RecvEidData()
115 : {
116 0 : recvEidData_.resize(sendEidData_.size());
117 0 : socket_->RecvAsync(reinterpret_cast<u8*>(recvEidData_.data()), recvEidData_.size());
118 0 : HCCL_INFO("[AicpuTsUboeChannel::%s] recv eid data, size=%llu", __func__, recvEidData_.size());
119 0 : }
120 :
121 0 : void AicpuTsUboeChannel::RecvEidDataProcess() { RmtEidUnpackProc(rmtAddr_); }
122 :
123 0 : void AicpuTsUboeChannel::RmtEidUnpackProc(Hccl::IpAddress& rmtAddr)
124 : {
125 0 : Hccl::IpAddress rmtEidAddr(recvEidData_);
126 0 : rmtAddr = rmtEidAddr;
127 0 : HCCL_INFO("[AicpuTsUboeChannel::%s] rmtAddr[%s]", __func__, rmtAddr.Describe().c_str());
128 0 : }
129 :
130 0 : void AicpuTsUboeChannel::SendFinish()
131 : {
132 0 : HCCL_INFO("start send Finish Msg [%s]", FINISH_MSG);
133 0 : sendFinishMsg_ = std::vector<char>(FINISH_MSG, FINISH_MSG + FINISH_MSG_SIZE);
134 0 : socket_->SendAsync(sendFinishMsg_.data(), FINISH_MSG_SIZE);
135 0 : HCCL_INFO("end send Finish Msg [%s]", FINISH_MSG);
136 0 : }
137 :
138 0 : void AicpuTsUboeChannel::RecvFinish()
139 : {
140 0 : recvFinishMsg_.resize(FINISH_MSG_SIZE);
141 0 : HCCL_INFO("start recv Finish Msg [%s]", FINISH_MSG);
142 0 : socket_->RecvAsync(reinterpret_cast<u8*>(recvFinishMsg_.data()), FINISH_MSG_SIZE);
143 0 : HCCL_INFO("end recv Finish Msg [%s]", FINISH_MSG);
144 0 : }
145 :
146 0 : void AicpuTsUboeChannel::HandleProcessData()
147 : {
148 0 : if (RecvDataProcess()) {
149 0 : uboeStatus = UboeStatus::SEND_FIN;
150 : } else {
151 0 : channelStatus = ChannelStatus::READY;
152 0 : uboeStatus = UboeStatus::READY;
153 : }
154 0 : }
155 :
156 2 : void AicpuTsUboeChannel::ProcessUboeState()
157 : {
158 2 : auto SetState = [this](UboeStatus next, ChannelStatus ch) {
159 2 : this->uboeStatus = next;
160 2 : this->channelStatus = ch;
161 4 : };
162 :
163 2 : switch (uboeStatus) {
164 2 : case UboeStatus::INIT:
165 2 : SetState(UboeStatus::SEND_EID, ChannelStatus::SOCKET_OK);
166 2 : break;
167 0 : case UboeStatus::SEND_EID:
168 0 : SendEidData();
169 0 : SetState(UboeStatus::RECV_EID, channelStatus);
170 0 : break;
171 0 : case UboeStatus::RECV_EID:
172 0 : RecvEidData();
173 0 : SetState(UboeStatus::PROCESS_EID_DATA, channelStatus);
174 0 : break;
175 0 : case UboeStatus::PROCESS_EID_DATA:
176 0 : RecvEidDataProcess();
177 0 : SetState(UboeStatus::BUILD_CONN, channelStatus);
178 0 : break;
179 0 : case UboeStatus::BUILD_CONN:
180 0 : BuildConn();
181 0 : SetState(UboeStatus::SEND_SIZE, channelStatus);
182 0 : break;
183 0 : case UboeStatus::SEND_SIZE:
184 0 : if (IsResReady()) {
185 0 : SendDataSize();
186 0 : SetState(UboeStatus::RECV_SIZE, channelStatus);
187 : }
188 0 : break;
189 0 : case UboeStatus::RECV_SIZE:
190 0 : RecvDataSize();
191 0 : SetState(isRecvFirst_ ? UboeStatus::RECV_DATA : UboeStatus::SEND_DATA, channelStatus);
192 0 : break;
193 0 : case UboeStatus::SEND_DATA:
194 0 : SendExchangeData();
195 0 : SetState(isRecvFirst_ ? UboeStatus::PROCESS_DATA : UboeStatus::RECV_DATA, channelStatus);
196 0 : break;
197 0 : case UboeStatus::RECV_DATA:
198 0 : RecvExchangeData();
199 0 : SetState(isRecvFirst_ ? UboeStatus::SEND_DATA : UboeStatus::PROCESS_DATA, channelStatus);
200 0 : break;
201 0 : case UboeStatus::PROCESS_DATA:
202 0 : HandleProcessData();
203 0 : break;
204 0 : case UboeStatus::SEND_FIN:
205 0 : if (IsConnsReady()) {
206 0 : SendFinish();
207 0 : SetState(UboeStatus::RECV_FIN, channelStatus);
208 : }
209 0 : break;
210 0 : case UboeStatus::RECV_FIN:
211 0 : RecvFinish();
212 0 : SetState(UboeStatus::SET_READY, channelStatus);
213 0 : break;
214 0 : case UboeStatus::SET_READY:
215 0 : channelStatus = ChannelStatus::READY;
216 0 : SetState(UboeStatus::READY, ChannelStatus::READY);
217 0 : break;
218 0 : default:
219 0 : break;
220 : }
221 2 : }
222 :
223 3 : HcclResult AicpuTsUboeChannel::CheckSocketStatus(const std::string& socketOperator)
224 : {
225 3 : CHK_PTR_NULL(socket_);
226 2 : auto timeout = std::chrono::seconds(Hccl::EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
227 2 : auto startTime = std::chrono::steady_clock::now();
228 2 : uint32_t retryCount = 0;
229 : while (true) {
230 2 : Hccl::SocketStatus socketStatus = socket_->GetAsyncStatus();
231 2 : if (socketStatus == Hccl::SocketStatus::OK) {
232 : auto elapsed
233 1 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
234 1 : .count();
235 1 : HCCL_INFO(
236 : "[AicpuTsUboeChannel][%s] socket operation[%s] success, elapsed[%lld]ms, retryCount[%u]", __func__,
237 : socketOperator.c_str(), elapsed, retryCount);
238 1 : break;
239 : }
240 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout || socketStatus == Hccl::SocketStatus::TIMEOUT) {
241 : auto elapsed
242 1 : = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - startTime)
243 1 : .count();
244 1 : HCCL_ERROR(
245 : "[AicpuTsUboeChannel][%s] socket operation[%s] timeout, socketStatus[%u], elapsed[%lld]ms, "
246 : "retryCount[%u]",
247 : __func__, socketOperator.c_str(), static_cast<uint32_t>(socketStatus), elapsed, retryCount);
248 1 : return HCCL_E_TIMEOUT;
249 : }
250 0 : retryCount++;
251 0 : }
252 1 : return HCCL_SUCCESS;
253 : }
254 :
255 5 : HcclResult AicpuTsUboeChannel::UpdateMemInfo(HcommMemHandle* memHandles, uint32_t memHandleNum)
256 : {
257 5 : std::vector<Hccl::LocalRmaBuffer*> bufferVecTemp;
258 5 : CHK_RET(MakeRmaBufferVecFromMemHandles(memHandles, memHandleNum, bufferVecTemp, "AicpuTsUboeChannel"));
259 :
260 4 : if (bufferVecTemp.size() == 0) {
261 1 : HCCL_WARNING("[AicpuTsUboeChannel][%s] bufferNum is 0.", __func__);
262 1 : return HCCL_SUCCESS;
263 : }
264 3 : CHK_PTR_NULL(socket_);
265 2 : HCCL_INFO("[AicpuTsUboeChannel][%s] bufferNum[%zu]", __func__, bufferVecTemp.size());
266 :
267 2 : std::vector<char> localSendData;
268 2 : Hccl::BinaryStream sendStream;
269 2 : BufferVecPack(sendStream, bufferVecTemp);
270 2 : sendStream.Dump(localSendData);
271 :
272 2 : u32 sendSize = localSendData.size();
273 2 : socket_->SendAsync(&sendSize, sizeof(sendSize));
274 2 : HCCL_INFO("[AicpuTsUboeChannel][%s] Send size[%u] of data.", __func__, sendSize);
275 4 : CHK_RET(CheckSocketStatus("SendDataSize"));
276 :
277 1 : u32 recvSize = 0;
278 1 : socket_->RecvAsync(reinterpret_cast<u8*>(&recvSize), sizeof(recvSize));
279 2 : CHK_RET(CheckSocketStatus("RecvDataSize"));
280 1 : HCCL_INFO("[AicpuTsUboeChannel][%s] Recv size[%u] of data.", __func__, recvSize);
281 :
282 1 : socket_->SendAsync(localSendData.data(), localSendData.size());
283 1 : HCCL_INFO("[AicpuTsUboeChannel][%s] Send data, size[%zu].", __func__, localSendData.size());
284 2 : CHK_RET(CheckSocketStatus("SendExchangeData"));
285 :
286 1 : std::vector<char> localRecvData(recvSize);
287 1 : socket_->RecvAsync(reinterpret_cast<u8*>(localRecvData.data()), localRecvData.size());
288 2 : CHK_RET(CheckSocketStatus("RecvExchangeData"));
289 1 : HCCL_INFO("[AicpuTsUboeChannel][%s] Recv data success.", __func__);
290 :
291 1 : std::vector<std::unique_ptr<Hccl::RemoteUbRmaBuffer>> rmtBufferTemp{};
292 1 : Hccl::BinaryStream recvStream(localRecvData);
293 1 : RmtBufferVecUnpackProc(static_cast<u32>(bufferVecTemp.size()), recvStream, rmtBufferTemp, UboeRmtBufType::BUFFER);
294 :
295 2 : rmtBufferVec_.insert(
296 1 : rmtBufferVec_.end(), std::make_move_iterator(rmtBufferTemp.begin()),
297 : std::make_move_iterator(rmtBufferTemp.end()));
298 1 : commonRes_.bufferVec.insert(commonRes_.bufferVec.end(), bufferVecTemp.begin(), bufferVecTemp.end());
299 1 : cacheValid_ = false;
300 1 : return HCCL_SUCCESS;
301 5 : }
302 :
303 2 : HcclResult AicpuTsUboeChannel::Clean()
304 : {
305 2 : memTransport_.reset();
306 :
307 2 : commonRes_.connVec.clear();
308 2 : connections_.clear();
309 :
310 2 : rmtNotifyVec_.clear();
311 2 : locBufferVec_.clear();
312 :
313 2 : recvData_.clear();
314 2 : recvFinishMsg_.clear();
315 2 : recvEidData_.clear();
316 2 : sendData_.clear();
317 2 : sendFinishMsg_.clear();
318 2 : sendEidData_.clear();
319 :
320 2 : bufferNum_ = 0;
321 2 : connNum_ = 0;
322 2 : recvDataSize_ = 0;
323 2 : locAddr_ = Hccl::IpAddress();
324 2 : rmtAddr_ = Hccl::IpAddress();
325 :
326 : {
327 2 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
328 2 : rmtBufferVec_.clear();
329 2 : cacheValid_ = false;
330 2 : remoteUserMems_.clear();
331 2 : memInfoCopies_.clear();
332 2 : memInfoPointers_.clear();
333 2 : }
334 :
335 2 : return HCCL_SUCCESS;
336 : }
337 :
338 1 : HcclResult AicpuTsUboeChannel::Resume()
339 : {
340 1 : channelStatus = ChannelStatus::INIT;
341 1 : uboeStatus = UboeStatus::INIT;
342 1 : HCCL_INFO("[AicpuTsUboeChannel][%s] reset uboe state machine to rebuild connection.", __func__);
343 1 : return HCCL_SUCCESS;
344 : }
345 :
346 2 : ChannelStatus AicpuTsUboeChannel::GetStatus()
347 : {
348 2 : if (channelStatus == ChannelStatus::READY) {
349 0 : return channelStatus;
350 : }
351 2 : if (channelStatus == ChannelStatus::INIT)
352 2 : uboeStatus = UboeStatus::INIT;
353 :
354 2 : if (!IsSocketReady())
355 0 : return channelStatus;
356 :
357 2 : ProcessUboeState();
358 :
359 2 : return channelStatus;
360 : }
361 :
362 : } // namespace hcomm
|
