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