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