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