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 "aicpu_ts_urma_channel.h"
12 : #include "endpoint.h"
13 : #include "orion_adpt_utils.h"
14 : #include "hcomm_c_adpt.h"
15 : #include "config_log.h"
16 : #include "endpoint.h"
17 :
18 : // Orion
19 : #include "adapter_rts_common.h"
20 : #include "topo_common_types.h"
21 : #include "virtual_topo.h"
22 : #include "aicpu_res_package_helper.h"
23 : #include "makebufs_helper.h"
24 :
25 : namespace hcomm {
26 : constexpr uint16_t DEFAULT_LISTENING_PORT = 60001;
27 :
28 10 : AicpuTsUrmaChannel::AicpuTsUrmaChannel(EndpointHandle endpointHandle, const HcommChannelDesc& channelDesc)
29 10 : : endpointHandle_(endpointHandle),
30 10 : channelDesc_(channelDesc)
31 10 : {}
32 :
33 11 : AicpuTsUrmaChannel::~AicpuTsUrmaChannel()
34 : {
35 10 : if (channelDesc_.socket == nullptr && socket_ != nullptr) {
36 0 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
37 0 : socket_ = nullptr;
38 : }
39 11 : }
40 :
41 3 : HcclResult AicpuTsUrmaChannel::ParseInputParam()
42 : {
43 : // 1. 从 endpointHandle_，获得 localEp_ 和 rdmaHandle_
44 : // TODO: 使用 HcommEndpointGet
45 3 : Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
46 3 : CHK_PTR_NULL(localEpPtr);
47 3 : localEp_ = localEpPtr->GetEndpointDesc();
48 3 : rdmaHandle_ = localEpPtr->GetRdmaHandle();
49 :
50 3 : HCCL_INFO("[AicpuTsUrmaChannel][%s] localProtocol[%d]", __func__, localEp_.protocol);
51 :
52 : // 2. 从 channelDesc_，获得 remoteEp_, socket_ 和 notifyNum
53 3 : remoteEp_ = channelDesc_.remoteEndpoint;
54 3 : socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
55 3 : notifyNum_ = channelDesc_.notifyNum;
56 3 : commonRes_.bufferVec.clear();
57 :
58 3 : if (channelDesc_.exchangeAllMems) {
59 : // 3. Get memHandles from endpoint
60 2 : HCCL_INFO("[AicpuTsUrmaChannel][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
61 2 : std::shared_ptr<Hccl::LocalUbRmaBuffer>* memHandles = nullptr;
62 2 : uint32_t memHandleNum = 0;
63 3 : CHK_RET(static_cast<HcclResult>(
64 : HcommMemGetAllMemHandles(endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
65 2 : HCCL_INFO("[AicpuTsUrmaChannel][%s] Got memHandleNum[%u].", __func__, memHandleNum);
66 3 : for (uint32_t i = 0; i < memHandleNum; ++i) {
67 2 : std::shared_ptr<Hccl::LocalUbRmaBuffer>& localUbRmaBuffer = memHandles[i];
68 2 : CHK_SMART_PTR_NULL(localUbRmaBuffer);
69 1 : Hccl::Buffer* buf = localUbRmaBuffer->GetBuf();
70 1 : CHK_PTR_NULL(buf);
71 1 : HCCL_INFO(
72 : "[AicpuTsUrmaChannel][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], type[%d], memInfo[%s].",
73 : __func__, i, static_cast<unsigned long long>(localUbRmaBuffer->GetAddr()),
74 : static_cast<unsigned long long>(localUbRmaBuffer->GetSize()), static_cast<int>(buf->GetMemType()),
75 : buf->GetMemInfo().c_str());
76 1 : commonRes_.bufferVec.push_back(localUbRmaBuffer.get());
77 : }
78 : } else {
79 : // 3. 从 channelDesc 的 memHandle，获得 bufs_
80 1 : HCCL_INFO("[AicpuTsUrmaChannel][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
81 1 : CHK_RET(MakeRmaBufferVecFromMemHandles(
82 : channelDesc_.memHandles, channelDesc_.memHandleNum, commonRes_.bufferVec, "AicpuTsUrmaChannel"));
83 : }
84 :
85 2 : return HCCL_SUCCESS;
86 : }
87 :
88 0 : HcclResult AicpuTsUrmaChannel::BuildAttr()
89 : {
90 0 : attr_.devicePhyId = localEp_.loc.device.devPhyId;
91 0 : attr_.opMode = Hccl::OpMode::OPBASE;
92 0 : return HCCL_SUCCESS;
93 : }
94 :
95 0 : HcclResult AicpuTsUrmaChannel::BuildConnection()
96 : {
97 0 : UbConnBuildContext ctx;
98 0 : CHK_RET(PrepareUbConnBuildContext(localEp_, remoteEp_, channelDesc_, ctx));
99 :
100 0 : Hccl::OpMode opMode = Hccl::OpMode::OPBASE;
101 0 : bool devUsed = true; // aicpu 为 true
102 0 : std::unique_ptr<Hccl::DevUbConnection> ubConn = nullptr;
103 0 : switch (ctx.protocol) {
104 0 : case Hccl::LinkProtocol::UB_TP:
105 0 : EXCEPTION_CATCH(
106 : ubConn = std::make_unique<Hccl::DevUbTpConnection>(
107 : rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, devUsed, Hccl::HrtUbJfcMode::STARS_POLL,
108 : Hccl::IpAddress(), Hccl::IpAddress(), ctx.qosPre, COMM_ENGINE_AICPU_TS, ctx.sqDepth),
109 : return HCCL_E_PTR);
110 0 : break;
111 0 : case Hccl::LinkProtocol::UB_CTP:
112 0 : EXCEPTION_CATCH(
113 : ubConn = std::make_unique<Hccl::DevUbCtpConnection>(
114 : rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, devUsed, Hccl::HrtUbJfcMode::STARS_POLL,
115 : Hccl::IpAddress(), Hccl::IpAddress(), ctx.qosPre, COMM_ENGINE_AICPU_TS, ctx.sqDepth),
116 : return HCCL_E_PTR);
117 0 : break;
118 0 : default:
119 0 : HCCL_ERROR("%s No LinkProtocol to match", __func__);
120 0 : break;
121 : }
122 0 : CHK_SMART_PTR_NULL(ubConn);
123 :
124 0 : if (devBaseAttr_.maxReadSize == 0 || devBaseAttr_.maxWriteSize == 0) {
125 0 : HCCL_ERROR(
126 : "%s maxReadSize[%u] or maxWriteSize[%u] must not be zero", __func__, devBaseAttr_.maxReadSize,
127 : devBaseAttr_.maxWriteSize);
128 0 : return HCCL_E_PARA;
129 : }
130 0 : ubConn->SetMaxReadSize(devBaseAttr_.maxReadSize);
131 0 : ubConn->SetMaxWriteSize(devBaseAttr_.maxWriteSize);
132 0 : HCCL_INFO("%s maxReadSize[%u], maxWriteSize[%u]", __func__, devBaseAttr_.maxReadSize, devBaseAttr_.maxWriteSize);
133 :
134 0 : commonRes_.connVec.clear();
135 0 : commonRes_.connVec.emplace_back(ubConn.get());
136 0 : connections_.clear();
137 0 : connections_.push_back(std::move(ubConn));
138 0 : return HCCL_SUCCESS;
139 0 : }
140 :
141 0 : HcclResult AicpuTsUrmaChannel::BuildNotify()
142 : {
143 0 : localNotifies_.clear();
144 0 : commonRes_.notifyVec.clear();
145 0 : bool devUsed = true;
146 0 : for (uint32_t i = 0; i < notifyNum_; ++i) {
147 0 : std::unique_ptr<Hccl::UbLocalNotify> notifyPtr = nullptr;
148 0 : EXCEPTION_CATCH(notifyPtr = std::make_unique<Hccl::UbLocalNotify>(rdmaHandle_, devUsed), return HCCL_E_PTR);
149 0 : commonRes_.notifyVec.push_back(notifyPtr.get());
150 0 : localNotifies_.push_back(std::move(notifyPtr));
151 0 : }
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : HcclResult AicpuTsUrmaChannel::BuildUbMemTransport()
156 : {
157 0 : Hccl::BaseMemTransport::LocCntNotifyRes locCntNotifyRes{};
158 0 : locCntNotifyRes.vec.clear();
159 0 : locCntNotifyRes.desc.clear();
160 0 : const Hccl::Socket& socket = *socket_;
161 :
162 0 : Hccl::LinkData linkData = BuildDefaultLinkData();
163 0 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
164 :
165 0 : bool isRecvFirst = socket.GetRole() == Hccl::SocketRole::CLIENT ? true : false;
166 :
167 : // make_unique / make_shared / release 包一层抛异常的宏
168 0 : EXCEPTION_CATCH(
169 : memTransport_ = std::make_unique<Hccl::UbMemTransport>(
170 : commonRes_, attr_, linkData, socket, rdmaHandle_, locCntNotifyRes, isRecvFirst),
171 : return HCCL_E_PTR);
172 0 : return HCCL_SUCCESS;
173 0 : }
174 :
175 0 : HcclResult AicpuTsUrmaChannel::BuildSocket()
176 : {
177 0 : if (socket_ != nullptr) {
178 0 : return HCCL_SUCCESS;
179 : }
180 0 : HCCL_INFO("[AicpuTsUrmaChannel][%s] socket ptr is NULL, rebuildSocket", __func__);
181 0 : Hccl::IpAddress ipaddr{};
182 0 : CHK_RET(CommAddrToIpAddress(localEp_.commAddr, ipaddr));
183 0 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::UB);
184 0 : Hccl::PortData localPort = Hccl::PortData(static_cast<Hccl::RankId>(localEp_.loc.device.devPhyId), type, 0, ipaddr);
185 0 : if (channelDesc_.role == HCOMM_SOCKET_ROLE_RESERVED) {
186 : Hccl::SocketHandle socketHandle
187 0 : = Hccl::SocketHandleManager::GetInstance().Create(localEp_.loc.device.devPhyId, localPort);
188 0 : EXCEPTION_CATCH(
189 : serverSocket_ = std::make_unique<Hccl::Socket>(
190 : socketHandle, ipaddr, DEFAULT_LISTENING_PORT, ipaddr, "server", Hccl::SocketRole::SERVER,
191 : Hccl::NicType::DEVICE_NIC_TYPE),
192 : return HCCL_E_PARA);
193 0 : HCCL_INFO("[AicpuTsUrmaChannel][%s] listen_socket_info[%s]", __func__, serverSocket_->Describe().c_str());
194 0 : EXCEPTION_CATCH(serverSocket_->Listen(), return HCCL_E_INTERNAL);
195 0 : Hccl::LinkData linkData = BuildDefaultLinkData();
196 0 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
197 0 : HCCL_INFO("[AicpuTsUrmaChannel][%s] built linkData: %s", __func__, linkData.Describe().c_str());
198 : std::string socketTag
199 0 : = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
200 0 : bool noRankId = true;
201 0 : Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, socketTag, noRankId);
202 0 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
203 0 : socketConfigHolder_ = std::make_unique<Hccl::SocketConfig>(socketConfig);
204 0 : socketConfig_ = socketConfigHolder_.get();
205 0 : } else {
206 0 : uint16_t port = channelDesc_.port;
207 0 : if (port == 0) {
208 0 : port = DEFAULT_LISTENING_PORT;
209 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
210 : }
211 0 : Hccl::LinkData linkData = BuildDefaultLinkData();
212 0 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
213 0 : HCCL_INFO("[AicpuTsUrmaChannel][%s] built linkData: %s", __func__, linkData.Describe().c_str());
214 : std::string socketTag
215 0 : = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
216 0 : bool isServer = (channelDesc_.role == HCOMM_SOCKET_ROLE_SERVER);
217 0 : Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, port, socketTag, isServer);
218 0 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
219 0 : socketConfigHolder_ = std::make_unique<Hccl::SocketConfig>(socketConfig);
220 0 : socketConfig_ = socketConfigHolder_.get();
221 0 : }
222 0 : return HCCL_SUCCESS;
223 : }
224 :
225 0 : HcclResult AicpuTsUrmaChannel::Init()
226 : {
227 : /*
228 : Argue result: make_unique 配合一场捕获的宏 EXCEPTION CATCH
229 : Attention: const 和引用
230 : */
231 : // TODO: 处理抛异常
232 : s32 devLogicId;
233 0 : CHK_RET(ParseInputParam());
234 0 : CHK_RET(hrtGetDevice(&devLogicId));
235 0 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
236 0 : CHK_RET(StartListen());
237 0 : CHK_RET(BuildSocket());
238 0 : CHK_RET(BuildAttr());
239 : /*
240 : HccpRaGetDevBaseAttr
241 : 获取urma read/write 单个wr的最大传输数据大小
242 : 调用前,rdmaHandle_要在ParseInputParam中被赋值好,之后BuildConnection会使用获取的属性
243 : */
244 0 : CHK_RET(HccpRaGetDevBaseAttr(rdmaHandle_, &devBaseAttr_));
245 0 : CHK_RET(BuildConnection());
246 0 : CHK_RET(BuildNotify());
247 0 : CHK_RET(BuildUbMemTransport());
248 :
249 0 : return HCCL_SUCCESS;
250 : }
251 :
252 0 : HcclResult AicpuTsUrmaChannel::GetNotifyNum(uint32_t* notifyNum) const
253 : {
254 0 : *notifyNum = this->notifyNum_;
255 0 : return HCCL_SUCCESS;
256 : }
257 :
258 0 : HcclResult AicpuTsUrmaChannel::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
259 : {
260 0 : return memTransport_->GetRemoteMems(memNum, remoteMem, memInfos);
261 : }
262 :
263 2 : ChannelStatus AicpuTsUrmaChannel::GetStatus()
264 : {
265 2 : ChannelStatus out = Channel::TransportStatusToChannelStatus(memTransport_->GetStatus());
266 :
267 2 : if (isFirstPrintChannelInfo_ && out == ChannelStatus::READY) {
268 2 : std::string channelInfo = "create channel info:channel handle[";
269 2 : channelInfo.append(std::to_string(reinterpret_cast<uint64_t>(this)));
270 2 : channelInfo.append("] ");
271 2 : HcclResult ret = memTransport_->Describe(channelInfo);
272 2 : if (ret != HCCL_SUCCESS) {
273 1 : HCCL_ERROR("[AicpuTsUrmaChannel][%s] Describe channel info failed, ret=%d", __func__, ret);
274 1 : out = ChannelStatus::FAILED;
275 : } else {
276 1 : channelInfo.append(" TA[RM]"); // 目前TA只支持RM
277 1 : HCCL_CONFIG_DEBUG(hccl::HCCL_RES, "%s", channelInfo.c_str());
278 : }
279 2 : isFirstPrintChannelInfo_ = false;
280 2 : }
281 2 : return out;
282 : }
283 :
284 0 : HcclResult SetModuleDataName(Hccl::ModuleData& module, const std::string& name)
285 : {
286 0 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
287 0 : if (ret != 0) {
288 0 : HCCL_ERROR("[SetModuleDataName] strcpy_s name %s failed", name.c_str());
289 0 : return HCCL_E_INTERNAL;
290 : }
291 :
292 0 : return HCCL_SUCCESS;
293 : }
294 :
295 0 : HcclResult AicpuTsUrmaChannel::PackOpData(std::vector<char>& data)
296 : {
297 0 : std::vector<Hccl::ModuleData> dataVec;
298 0 : dataVec.resize(Hccl::AicpuResMgrType::__COUNT__);
299 :
300 0 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
301 0 : CHK_RET(SetModuleDataName(dataVec[resType], "UbMemTransport"));
302 :
303 0 : std::vector<char> result;
304 0 : Hccl::BinaryStream binaryStream;
305 0 : binaryStream << memTransport_->GetUniqueIdV2();
306 :
307 0 : binaryStream.Dump(result);
308 :
309 0 : dataVec[resType].data = result;
310 :
311 : Hccl::AicpuResPackageHelper helper;
312 0 : data = helper.GetPackedData(dataVec);
313 :
314 0 : return HCCL_SUCCESS;
315 0 : }
316 :
317 0 : HcclResult AicpuTsUrmaChannel::H2DResPack(std::vector<char>& buffer)
318 : {
319 0 : CHK_RET(PackOpData(buffer));
320 0 : HCCL_INFO(
321 : "[AicpuTsUrmaChannel][%s] Pack Buffer data[%p], Pack Buffer size[%zu].", __func__, buffer.data(),
322 : buffer.size());
323 0 : return HCCL_SUCCESS;
324 : }
325 :
326 1 : HcclResult AicpuTsUrmaChannel::Clean()
327 : {
328 1 : memTransport_.reset();
329 1 : return HCCL_SUCCESS;
330 : }
331 :
332 1 : HcclResult AicpuTsUrmaChannel::Resume()
333 : {
334 1 : BuildSocket();
335 1 : BuildConnection();
336 1 : BuildUbMemTransport();
337 1 : return HCCL_SUCCESS;
338 : }
339 :
340 2 : HcclResult AicpuTsUrmaChannel::UpdateMemInfo(HcommMemHandle* memHandles, uint32_t memHandleNum)
341 : {
342 2 : std::vector<Hccl::LocalRmaBuffer*> bufferVecTemp;
343 2 : CHK_RET(MakeRmaBufferVecFromMemHandles(memHandles, memHandleNum, bufferVecTemp, "AicpuTsUrmaChannel"));
344 1 : CHK_RET(memTransport_->UpdateMemInfo(bufferVecTemp));
345 1 : commonRes_.bufferVec.insert(commonRes_.bufferVec.end(), bufferVecTemp.begin(), bufferVecTemp.end());
346 1 : return HCCL_SUCCESS;
347 2 : }
348 :
349 : // 返回当前 channel 类型，供上层区分不同 channel 的能力和行为
350 0 : HcommChannelKind AicpuTsUrmaChannel::GetChannelKind() const { return HcommChannelKind::AICPU_TS_URMA; }
351 :
352 0 : HcclResult AicpuTsUrmaChannel::NotifyRecord([[maybe_unused]] const uint32_t remoteNotifyIdx)
353 : {
354 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] not supported yet.", __func__);
355 0 : return HCCL_E_NOT_SUPPORT;
356 : }
357 :
358 : HcclResult
359 0 : AicpuTsUrmaChannel::NotifyWait([[maybe_unused]] const uint32_t localNotifyIdx, [[maybe_unused]] const uint32_t timeout)
360 : {
361 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] not supported yet.", __func__);
362 0 : return HCCL_E_NOT_SUPPORT;
363 : }
364 :
365 0 : HcclResult AicpuTsUrmaChannel::WriteWithNotify(
366 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] const uint64_t len,
367 : [[maybe_unused]] uint32_t remoteNotifyIdx)
368 : {
369 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] not supported yet.", __func__);
370 0 : return HCCL_E_NOT_SUPPORT;
371 : }
372 :
373 : HcclResult
374 0 : AicpuTsUrmaChannel::Write([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
375 : {
376 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] not supported yet.", __func__);
377 0 : return HCCL_E_NOT_SUPPORT;
378 : }
379 :
380 : HcclResult
381 0 : AicpuTsUrmaChannel::Read([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
382 : {
383 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] not supported yet.", __func__);
384 0 : return HCCL_E_NOT_SUPPORT;
385 : }
386 :
387 0 : HcclResult AicpuTsUrmaChannel::ChannelFence()
388 : {
389 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] not supported yet.", __func__);
390 0 : return HCCL_E_NOT_SUPPORT;
391 : }
392 :
393 1 : HcclResult AicpuTsUrmaChannel::StartListen()
394 : {
395 1 : if (channelDesc_.role != HCOMM_SOCKET_ROLE_SERVER) {
396 1 : return HCCL_SUCCESS;
397 : }
398 :
399 0 : uint16_t port = channelDesc_.port;
400 0 : HCCL_INFO(
401 : "[AicpuTsUrmaChannel::%s] Start. EndpointHandle[0x%llx], port[%u]", __func__,
402 : reinterpret_cast<uint64_t>(endpointHandle_), port);
403 0 : if (port == 0) {
404 0 : port = DEFAULT_LISTENING_PORT;
405 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
406 : }
407 0 : CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(endpointHandle_, port, nullptr)));
408 0 : HCCL_INFO("[AicpuTsUrmaChannel::%s] SUCCESS. port[%u].", __func__, port);
409 0 : return HCCL_SUCCESS;
410 : }
411 :
412 : } // namespace hcomm
|
