Line data Source code
1 : /**
2 : * Copyright (c) 2025 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 "ccu_urma_channel.h"
12 :
13 : #include "hcomm_c_adpt.h"
14 :
15 : #include "orion_adpt_utils.h"
16 :
17 : #include "exception_handler.h"
18 : #include "comm_mems.h"
19 :
20 : #include "config_log.h"
21 :
22 : // 暂时引入orion
23 : #include "local_ub_rma_buffer.h"
24 :
25 : namespace hcomm {
26 :
27 44 : CcuUrmaChannel::CcuUrmaChannel(const EndpointHandle locEndpointHandle, const HcommChannelDesc& channelDesc)
28 44 : : locEndpointHandle_(locEndpointHandle),
29 44 : channelDesc_(channelDesc)
30 44 : {}
31 :
32 16 : HcclResult BuildBufferInfos(
33 : HcommMemHandle* memHandles, uint32_t memHandleNum, std::vector<CcuTransport::CclBufferInfo>& bufferInfos)
34 : {
35 32 : for (uint32_t i = 0; i < memHandleNum; ++i) {
36 16 : auto localRmaBuffer = reinterpret_cast<Hccl::LocalUbRmaBuffer*>(memHandles[i]);
37 16 : CHK_PTR_NULL(localRmaBuffer);
38 16 : auto buf = localRmaBuffer->GetBuf();
39 16 : CHK_PTR_NULL(buf);
40 16 : HCCL_INFO("[BuildBufferInfos] localRmaBuffer[%s]", localRmaBuffer->Describe().c_str());
41 :
42 16 : std::array<char, HCCL_RES_TAG_MAX_LEN> memInfo{};
43 16 : std::string tag = buf->GetMemInfo();
44 16 : if (UNLIKELY(tag.size() >= HCCL_RES_TAG_MAX_LEN)) {
45 0 : HCCL_ERROR("[BuildBufferInfos] tagSize exceeds limit[%u]", HCCL_RES_TAG_MAX_LEN);
46 0 : return HCCL_E_PARA;
47 : }
48 48 : CHK_SAFETY_FUNC_RET(memcpy_s(memInfo.data(), memInfo.size(), tag.c_str(), tag.size()));
49 16 : bufferInfos.emplace_back(
50 16 : localRmaBuffer->GetAddr(), static_cast<uint32_t>(localRmaBuffer->GetSize()), localRmaBuffer->GetTokenId(),
51 16 : localRmaBuffer->GetTokenValue(), hccl::ConvertHcclToCommMemType(buf->GetMemType()), memInfo);
52 16 : }
53 16 : return HCCL_SUCCESS;
54 : }
55 :
56 15 : static HcclResult CreateCcuTransport(
57 : UrmaEndpoint* ccuEndpoint, const Hccl::LinkData& linkData, Hccl::Socket* socket, HcommMemHandle* memHandles,
58 : uint32_t memHandleNum, uint32_t qos, uint32_t sqSize, std::unique_ptr<CcuTransport>& impl)
59 : {
60 15 : HCCL_INFO("[CcuUrmaChannel][%s] begin, sqSize[%u]", __func__, sqSize);
61 : // 当前ccu channel不支持按需申请cke
62 15 : CHK_PTR_NULL(ccuEndpoint);
63 15 : CHK_PTR_NULL(socket);
64 15 : CHK_PTR_NULL(memHandles);
65 :
66 15 : auto ret = HcclResult::HCCL_SUCCESS;
67 15 : auto* channelCtxPool = ccuEndpoint->GetCcuChannelCtxPool();
68 15 : CHK_PTR_NULL(channelCtxPool);
69 : // 申请ccu channel ctx， jetty ctx，wqebb，可能资源不足，需要回退
70 30 : ret = channelCtxPool->PrepareCreate({linkData}, sqSize);
71 15 : if (ret == HCCL_E_UNAVAIL) {
72 0 : HCCL_WARNING(
73 : "[CcuUrmaChannel][%s] prepare ccu channel ctx failed, "
74 : "ccu resources unavailable.",
75 : __func__);
76 0 : return ret;
77 : }
78 15 : CHK_RET(ret);
79 :
80 15 : CcuChannelCtxPool::CcuChannelCtx channelCtx{};
81 15 : CHK_RET(channelCtxPool->GetChannelCtx(linkData, channelCtx));
82 15 : const auto& channelInfo = channelCtx.first;
83 15 : const auto& ccuJettys = channelCtx.second;
84 :
85 15 : const auto& locAddr_ = linkData.GetLocalAddr();
86 15 : const auto& rmtAddr_ = linkData.GetRemoteAddr();
87 :
88 15 : CommAddr locAddr{}, rmtAddr{};
89 15 : CHK_RET(IpAddressToCommAddr(locAddr_, locAddr));
90 15 : CHK_RET(IpAddressToCommAddr(rmtAddr_, rmtAddr));
91 :
92 15 : CcuTransport::CcuConnectionType type_ = linkData.GetLinkProtocol() == Hccl::LinkProtocol::UB_CTP ?
93 : CcuTransport::CcuConnectionType::UB_CTP :
94 15 : CcuTransport::CcuConnectionType::UBC_TP;
95 :
96 15 : CcuTransport::CcuConnectionInfo connectionInfo{type_, locAddr, rmtAddr, channelInfo, ccuJettys, qos};
97 :
98 15 : std::vector<CcuTransport::CclBufferInfo> bufferInfos{};
99 15 : CHK_RET(BuildBufferInfos(memHandles, memHandleNum, bufferInfos));
100 :
101 : // 调用底层的创建函数 (CcuCreateTransport 通常是全局函数或静态函数)
102 : // 申请 xn cke可能失败，需要回退
103 15 : ret = CcuCreateTransport(socket, connectionInfo, bufferInfos, impl);
104 15 : if (ret == HCCL_E_UNAVAIL) {
105 0 : HCCL_WARNING("[CcuUrmaChannel][%s] failed, ccu resources unavailable.", __func__);
106 0 : return ret;
107 : }
108 15 : CHK_RET(ret);
109 :
110 15 : HCCL_INFO("[CcuUrmaChannel][%s] end, transport created.", __func__);
111 15 : return HCCL_SUCCESS;
112 15 : }
113 :
114 15 : static HcclResult CheckEndpointDesc(const EndpointDesc& locDesc, const EndpointDesc& rmtDesc)
115 : {
116 15 : if (locDesc.protocol != rmtDesc.protocol) {
117 0 : HCCL_ERROR(
118 : "[CcuUrmaChannel][%s] failed, endpoints protocols are not same, "
119 : "loc[%d] rmt[%d].",
120 : __func__, locDesc.protocol, rmtDesc.protocol);
121 0 : return HcclResult::HCCL_E_PARA;
122 : }
123 :
124 15 : if (locDesc.protocol != COMM_PROTOCOL_UB_CTP && locDesc.protocol != COMM_PROTOCOL_UBC_TP) {
125 0 : HCCL_ERROR("[CcuUrmaChannel][%s] failed, protocol[%d] are not supported in ccu.", __func__, locDesc.protocol);
126 0 : return HcclResult::HCCL_E_PARA;
127 : }
128 :
129 15 : return HcclResult::HCCL_SUCCESS;
130 : }
131 :
132 15 : HcclResult CcuUrmaChannel::Init()
133 : {
134 : EXCEPTION_HANDLE_BEGIN
135 15 : CHK_PTR_NULL(channelDesc_.socket);
136 15 : auto* socket = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
137 : // 当前socket在外部统一触发connect，建议之后改为异步建链流程内触发
138 :
139 15 : CHK_PTR_NULL(locEndpointHandle_);
140 15 : void* endpoint{nullptr};
141 15 : CHK_RET(static_cast<HcclResult>(HcommEndpointGet(locEndpointHandle_, &endpoint)));
142 15 : UrmaEndpoint* ccuEndpoint = dynamic_cast<UrmaEndpoint*>(static_cast<Endpoint*>(endpoint));
143 15 : CHK_PTR_NULL(ccuEndpoint);
144 15 : const auto& locEndpointDesc = ccuEndpoint->GetEndpointDesc();
145 :
146 15 : CHK_RET(CheckEndpointDesc(locEndpointDesc, channelDesc_.remoteEndpoint));
147 :
148 15 : auto linkData = BuildDefaultLinkData();
149 15 : CHK_RET(EndpointDescPairToLinkData(locEndpointDesc, channelDesc_.remoteEndpoint, linkData));
150 :
151 15 : if (channelDesc_.memHandleNum == 0) {
152 0 : HCCL_ERROR("[CcuUrmaChannel][%s] failed, unsupported memHandleNum[%u].", __func__, channelDesc_.memHandleNum);
153 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
154 : }
155 15 : CHK_PTR_NULL(channelDesc_.memHandles);
156 :
157 : // 当前建链不支持资源扩容，CCU资源默认固定为8
158 15 : HCCL_WARNING("[CcuUrmaChannel][%s] now only support notify num is 8.", __func__);
159 15 : HCCL_WARNING("[CcuUrmaChannel][%s] now only support to exchange hccl buffer.", __func__);
160 15 : CHK_RET_UNAVAIL(CreateCcuTransport(
161 : ccuEndpoint, linkData, socket, channelDesc_.memHandles, channelDesc_.memHandleNum, channelDesc_.qos,
162 : channelDesc_.ubAttr.sqDepth, impl_));
163 :
164 0 : EXCEPTION_HANDLE_END
165 15 : return HCCL_SUCCESS;
166 : }
167 :
168 19 : ChannelStatus CcuUrmaChannel::GetStatus()
169 : {
170 19 : if (!impl_) {
171 0 : HCCL_ERROR("[CcuUrmaChannel][%s] failed, impl is nullptr.", __func__);
172 0 : return ChannelStatus::FAILED;
173 : }
174 :
175 19 : CcuTransport::TransStatus status = impl_->GetStatus();
176 19 : ChannelStatus out = ChannelStatus::INIT;
177 19 : switch (status) {
178 2 : case CcuTransport::TransStatus::READY:
179 2 : out = ChannelStatus::READY;
180 2 : break;
181 1 : case CcuTransport::TransStatus::SOCKET_TIMEOUT:
182 1 : HCCL_ERROR("[CcuUrmaChannel][%s] error status[%s].", __func__, status.Describe().c_str());
183 1 : out = ChannelStatus::SOCKET_TIMEOUT;
184 1 : break;
185 1 : case CcuTransport::TransStatus::CONNECT_FAILED:
186 1 : HCCL_ERROR("[CcuUrmaChannel][%s] error status[%s].", __func__, status.Describe().c_str());
187 1 : out = ChannelStatus::FAILED;
188 1 : break;
189 15 : default:
190 15 : break;
191 : }
192 :
193 19 : if (isFirstPrintChannelInfo_ && out == ChannelStatus::READY) {
194 2 : std::string channelInfo = "create channel info:channel handle[";
195 2 : channelInfo.append(std::to_string(reinterpret_cast<uint64_t>(this)));
196 2 : channelInfo.append("] ");
197 2 : HcclResult ret = impl_->Describe(channelInfo);
198 2 : if (ret != HCCL_SUCCESS) {
199 1 : HCCL_ERROR("[CcuUrmaChannel][%s] Describe channel info failed, ret=%d", __func__, ret);
200 1 : out = ChannelStatus::FAILED;
201 : } else {
202 1 : channelInfo.append(" TA[RM]"); // 目前TA只支持RM
203 1 : HCCL_CONFIG_DEBUG(hccl::HCCL_RES, "%s", channelInfo.c_str());
204 : }
205 2 : isFirstPrintChannelInfo_ = false;
206 2 : }
207 19 : return out; // todo: AICPU 重新定义基类的状态后，需要修改为CONNECTING
208 : }
209 :
210 17 : uint32_t CcuUrmaChannel::GetDieId() const
211 : {
212 17 : if (!impl_) {
213 0 : return UINT32_MAX;
214 : }
215 :
216 17 : return impl_->GetDieId();
217 : }
218 :
219 36 : uint32_t CcuUrmaChannel::GetChannelId() const
220 : {
221 36 : if (!impl_) {
222 3 : return UINT32_MAX;
223 : }
224 33 : return impl_->GetChannelId();
225 : }
226 :
227 8 : HcclResult CcuUrmaChannel::GetRmtSignalAddrByIndex(uint32_t index, uint64_t& rmtCkeAddr) const
228 : {
229 8 : CHK_PTR_NULL(impl_);
230 8 : CHK_RET(impl_->GetRmtSignalAddrByIndex(index, rmtCkeAddr));
231 8 : return HcclResult::HCCL_SUCCESS;
232 : }
233 :
234 6 : HcclResult CcuUrmaChannel::GetRmtVarAddrByIndex(uint32_t index, uint64_t& rmtXnAddr) const
235 : {
236 6 : CHK_PTR_NULL(impl_);
237 6 : CHK_RET(impl_->GetRmtVarAddrByIndex(index, rmtXnAddr));
238 6 : return HcclResult::HCCL_SUCCESS;
239 : }
240 :
241 8 : HcclResult CcuUrmaChannel::GetRmtCcuBufferTokenInfo(uint32_t& rmtTokenId, uint32_t& rmtTokenValue) const
242 : {
243 8 : CHK_PTR_NULL(impl_);
244 8 : CHK_RET(impl_->GetRmtCcuBufferTokenInfo(rmtTokenId, rmtTokenValue));
245 8 : return HcclResult::HCCL_SUCCESS;
246 : }
247 :
248 12 : HcclResult CcuUrmaChannel::GetLocCkeByIndex(const uint32_t index, uint32_t& locCkeId) const
249 : {
250 12 : CHK_PTR_NULL(impl_);
251 12 : CHK_RET(impl_->GetLocCkeByIndex(index, locCkeId));
252 12 : return HcclResult::HCCL_SUCCESS;
253 : }
254 :
255 10 : HcclResult CcuUrmaChannel::GetLocXnByIndex(const uint32_t index, uint32_t& locXnId) const
256 : {
257 10 : CHK_PTR_NULL(impl_);
258 10 : CHK_RET(impl_->GetLocXnByIndex(index, locXnId));
259 10 : return HcclResult::HCCL_SUCCESS;
260 : }
261 :
262 7 : HcclResult CcuUrmaChannel::GetRmtCkeByIndex(const uint32_t index, uint32_t& rmtCkeId) const
263 : {
264 7 : CHK_PTR_NULL(impl_);
265 7 : CHK_RET(impl_->GetRmtCkeByIndex(index, rmtCkeId));
266 7 : return HcclResult::HCCL_SUCCESS;
267 : }
268 :
269 5 : HcclResult CcuUrmaChannel::GetRmtXnByIndex(const uint32_t index, uint32_t& rmtXnId) const
270 : {
271 5 : CHK_PTR_NULL(impl_);
272 5 : CHK_RET(impl_->GetRmtXnByIndex(index, rmtXnId));
273 5 : return HcclResult::HCCL_SUCCESS;
274 : }
275 :
276 0 : HcclResult CcuUrmaChannel::GetRmtWishCntXnAddr(const std::string& resGroupTag, uint64_t& wishCntXnAddr) const
277 : {
278 0 : CHK_PTR_NULL(impl_);
279 0 : CHK_RET(impl_->GetRmtWishCntXnAddr(resGroupTag, wishCntXnAddr));
280 0 : return HcclResult::HCCL_SUCCESS;
281 : }
282 :
283 0 : HcclResult CcuUrmaChannel::GetRmtBuffer(uint64_t& addr, uint32_t& size, uint32_t& tokenId, uint32_t& tokenValue) const
284 : {
285 0 : CHK_PTR_NULL(impl_);
286 0 : CcuTransport::CclBufferInfo bufInfo{};
287 0 : constexpr uint32_t bufNum = 0; // 当前不支持
288 0 : CHK_RET(impl_->GetRmtBuffer(bufInfo, bufNum));
289 :
290 0 : addr = bufInfo.addr;
291 0 : size = bufInfo.size;
292 0 : tokenId = bufInfo.tokenId;
293 0 : tokenValue = bufInfo.tokenValue;
294 0 : return HcclResult::HCCL_SUCCESS;
295 : }
296 :
297 0 : HcclResult CcuUrmaChannel::GetNotifyNum(uint32_t* notifyNum) const
298 : {
299 0 : CHK_PTR_NULL(impl_);
300 0 : CHK_RET(impl_->GetCkeNum(*notifyNum));
301 0 : return HcclResult::HCCL_SUCCESS;
302 : }
303 :
304 0 : HcclResult CcuUrmaChannel::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
305 : {
306 0 : CHK_PTR_NULL(impl_);
307 0 : return impl_->GetRemoteMems(memNum, remoteMem, memInfos);
308 : }
309 :
310 2 : HcclResult CcuUrmaChannel::Clean()
311 : {
312 2 : CHK_PTR_NULL(impl_);
313 1 : impl_->Clean();
314 1 : return HcclResult::HCCL_SUCCESS;
315 : }
316 :
317 1 : HcclResult CcuUrmaChannel::Resume() { return HCCL_SUCCESS; }
318 :
319 0 : HcclResult CcuUrmaChannel::UpdateMemInfo(HcommMemHandle* memHandles, uint32_t memHandleNum)
320 : {
321 0 : std::vector<CcuTransport::CclBufferInfo> bufferVecTemp{};
322 0 : CHK_RET(BuildBufferInfos(memHandles, memHandleNum, bufferVecTemp));
323 0 : return impl_->UpdateMemInfo(bufferVecTemp);
324 0 : }
325 :
326 0 : HcclResult CcuUrmaChannel::NotifyRecord([[maybe_unused]] const uint32_t remoteNotifyIdx)
327 : {
328 0 : HCCL_INFO("[CcuUrmaChannel::%s] not supported yet.", __func__);
329 0 : return HCCL_E_NOT_SUPPORT;
330 : }
331 :
332 : HcclResult
333 0 : CcuUrmaChannel::NotifyWait([[maybe_unused]] const uint32_t localNotifyIdx, [[maybe_unused]] const uint32_t timeout)
334 : {
335 0 : HCCL_INFO("[CcuUrmaChannel::%s] not supported yet.", __func__);
336 0 : return HCCL_E_NOT_SUPPORT;
337 : }
338 :
339 0 : HcclResult CcuUrmaChannel::WriteWithNotify(
340 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] const uint64_t len,
341 : [[maybe_unused]] uint32_t remoteNotifyIdx)
342 : {
343 0 : HCCL_INFO("[CcuUrmaChannel::%s] not supported yet.", __func__);
344 0 : return HCCL_E_NOT_SUPPORT;
345 : }
346 :
347 : HcclResult
348 0 : CcuUrmaChannel::Write([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
349 : {
350 0 : HCCL_INFO("[CcuUrmaChannel::%s] not supported yet.", __func__);
351 0 : return HCCL_E_NOT_SUPPORT;
352 : }
353 :
354 : HcclResult
355 0 : CcuUrmaChannel::Read([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
356 : {
357 0 : HCCL_INFO("[CcuUrmaChannel::%s] not supported yet.", __func__);
358 0 : return HCCL_E_NOT_SUPPORT;
359 : }
360 :
361 0 : HcclResult CcuUrmaChannel::ChannelFence()
362 : {
363 0 : HCCL_INFO("[CcuUrmaChannel::%s] not supported yet.", __func__);
364 0 : return HCCL_E_NOT_SUPPORT;
365 : }
366 :
367 : } // namespace hcomm
|
