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