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_p2p_channel.h"
12 : #include "endpoint.h"
13 : #include "orion_adpt_utils.h"
14 : #include "hcomm_c_adpt.h"
15 : #include "adapter_rts_common.h"
16 :
17 : #include "virtual_topo.h"
18 : #include "p2p_connection.h"
19 : #include "makebufs_helper.h"
20 :
21 : namespace hcomm {
22 :
23 2 : AicpuTsP2pChannel::AicpuTsP2pChannel(EndpointHandle endpointHandle, const HcommChannelDesc& channelDesc)
24 2 : : endpointHandle_(endpointHandle),
25 2 : channelDesc_(channelDesc)
26 2 : {}
27 :
28 2 : AicpuTsP2pChannel::~AicpuTsP2pChannel()
29 : {
30 2 : if (channelDesc_.socket == nullptr && socket_ != nullptr) {
31 0 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
32 0 : socket_ = nullptr;
33 : }
34 2 : }
35 :
36 2 : HcclResult AicpuTsP2pChannel::ParseInputParam()
37 : {
38 2 : Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
39 2 : CHK_PTR_NULL(localEpPtr);
40 2 : localEp_ = localEpPtr->GetEndpointDesc();
41 :
42 2 : HCCL_INFO("[AicpuTsP2pChannel][%s] localProtocol[%d]", __func__, localEp_.protocol);
43 :
44 2 : remoteEp_ = channelDesc_.remoteEndpoint;
45 2 : socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
46 2 : notifyNum_ = channelDesc_.notifyNum;
47 2 : commonRes_.bufferVec.clear();
48 :
49 2 : if (channelDesc_.exchangeAllMems) {
50 1 : HCCL_INFO("[AicpuTsP2pChannel][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
51 1 : std::shared_ptr<Hccl::LocalIpcRmaBuffer>* memHandles = nullptr;
52 1 : uint32_t memHandleNum = 0;
53 1 : CHK_RET(static_cast<HcclResult>(
54 : HcommMemGetAllMemHandles(endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
55 1 : HCCL_INFO("[AicpuTsP2pChannel][%s] Got memHandleNum[%u].", __func__, memHandleNum);
56 2 : for (uint32_t i = 0; i < memHandleNum; ++i) {
57 1 : std::shared_ptr<Hccl::LocalIpcRmaBuffer>& localIpcRmaBuffer = memHandles[i];
58 1 : CHK_SMART_PTR_NULL(localIpcRmaBuffer);
59 1 : auto buf = localIpcRmaBuffer->GetBuf();
60 1 : CHK_PTR_NULL(buf);
61 1 : HCCL_INFO(
62 : "[AicpuTsP2pChannel][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], "
63 : "memType[%d], memInfo[%s].",
64 : __func__, i, static_cast<unsigned long long>(buf->GetAddr()),
65 : static_cast<unsigned long long>(buf->GetSize()), static_cast<int>(buf->GetMemType()),
66 : buf->GetMemInfo().c_str());
67 1 : commonRes_.bufferVec.push_back(localIpcRmaBuffer.get());
68 : }
69 : } else {
70 1 : HCCL_INFO("[AicpuTsP2pChannel][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
71 1 : CHK_RET(MakeRmaBufferVecFromMemHandles(
72 : channelDesc_.memHandles, channelDesc_.memHandleNum, commonRes_.bufferVec, "AicpuTsP2pChannel"));
73 : }
74 :
75 2 : return HCCL_SUCCESS;
76 : }
77 :
78 0 : HcclResult AicpuTsP2pChannel::BuildAttr()
79 : {
80 0 : attr_.devicePhyId = localEp_.loc.device.devPhyId;
81 0 : attr_.opMode = Hccl::OpMode::OPBASE;
82 0 : return HCCL_SUCCESS;
83 : }
84 :
85 0 : HcclResult AicpuTsP2pChannel::BuildConnection()
86 : {
87 0 : std::unique_ptr<Hccl::P2PConnection> p2pConn = nullptr;
88 0 : std::string connTag = "P2P_CHANNEL_" + std::to_string(localEp_.loc.device.devPhyId);
89 :
90 0 : EXCEPTION_CATCH(p2pConn = std::make_unique<Hccl::P2PConnection>(socket_, connTag), return HCCL_E_PTR);
91 0 : CHK_SMART_PTR_NULL(p2pConn);
92 :
93 0 : commonRes_.connVec.clear();
94 0 : commonRes_.connVec.emplace_back(p2pConn.get());
95 0 : connections_.clear();
96 0 : connections_.push_back(std::move(p2pConn));
97 :
98 0 : return HCCL_SUCCESS;
99 0 : }
100 :
101 0 : HcclResult AicpuTsP2pChannel::BuildNotify()
102 : {
103 0 : localNotifies_.clear();
104 0 : commonRes_.notifyVec.clear();
105 0 : bool devUsed = true;
106 0 : for (uint32_t i = 0; i < notifyNum_; ++i) {
107 0 : std::unique_ptr<Hccl::IpcLocalNotify> notifyPtr = nullptr;
108 0 : EXCEPTION_CATCH(notifyPtr = std::make_unique<Hccl::IpcLocalNotify>(devUsed), return HCCL_E_PTR);
109 0 : commonRes_.notifyVec.push_back(notifyPtr.get());
110 0 : localNotifies_.push_back(std::move(notifyPtr));
111 0 : }
112 0 : return HCCL_SUCCESS;
113 : }
114 :
115 0 : HcclResult AicpuTsP2pChannel::BuildP2pMemTransport()
116 : {
117 0 : const Hccl::Socket& socket = *socket_;
118 :
119 0 : Hccl::LinkData linkData = BuildDefaultLinkData();
120 0 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
121 :
122 0 : EXCEPTION_CATCH(
123 : memTransport_ = std::make_unique<Hccl::P2PTransport>(commonRes_, attr_, linkData, socket), return HCCL_E_PTR);
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 0 : HcclResult AicpuTsP2pChannel::BuildSocket()
128 : {
129 0 : if (socket_ != nullptr) {
130 0 : return HCCL_SUCCESS;
131 : }
132 0 : HCCL_INFO("[AicpuTsP2pChannel][%s] socket ptr is NULL, rebuildSocket", __func__);
133 :
134 0 : Hccl::IpAddress ipaddr{};
135 0 : CHK_RET(CommAddrToIpAddress(localEp_.commAddr, ipaddr));
136 0 : Hccl::DevNetPortType type = Hccl::DevNetPortType(Hccl::ConnectProtoType::PCIE); // TODO PROTOTYPE P2P?
137 0 : Hccl::PortData localPort = Hccl::PortData(static_cast<Hccl::RankId>(localEp_.loc.device.devPhyId), type, 0, ipaddr);
138 : Hccl::SocketHandle socketHandle
139 0 : = Hccl::SocketHandleManager::GetInstance().Create(localEp_.loc.device.devPhyId, localPort);
140 0 : EXCEPTION_CATCH(
141 : serverSocket_ = std::make_unique<Hccl::Socket>(
142 : socketHandle, ipaddr, 60001, ipaddr, "server", Hccl::SocketRole::SERVER, Hccl::NicType::DEVICE_NIC_TYPE),
143 : return HCCL_E_PARA);
144 0 : HCCL_INFO("[AicpuTsP2pChannel][%s] listen_socket_info[%s]", __func__, serverSocket_->Describe().c_str());
145 0 : EXCEPTION_CATCH(serverSocket_->Listen(), return HCCL_E_INTERNAL);
146 :
147 0 : Hccl::LinkData linkData = BuildDefaultLinkData();
148 0 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
149 0 : HCCL_INFO("[AicpuTsP2pChannel][%s] built linkData: %s", __func__, linkData.Describe().c_str());
150 : std::string socketTag
151 0 : = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
152 0 : bool noRankId = true;
153 0 : Hccl::SocketConfig socketConfig = Hccl::SocketConfig(linkData, socketTag, noRankId);
154 0 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
155 :
156 0 : return HCCL_SUCCESS;
157 0 : }
158 :
159 0 : HcclResult AicpuTsP2pChannel::Init()
160 : {
161 0 : CHK_RET(ParseInputParam());
162 : s32 devLogicId;
163 0 : CHK_RET(hrtGetDevice(&devLogicId));
164 0 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
165 0 : CHK_RET(BuildSocket());
166 0 : CHK_RET(BuildAttr());
167 0 : CHK_RET(BuildConnection());
168 0 : CHK_RET(BuildNotify());
169 0 : CHK_RET(BuildP2pMemTransport());
170 :
171 0 : return HCCL_SUCCESS;
172 : }
173 :
174 0 : HcclResult AicpuTsP2pChannel::GetNotifyNum(uint32_t* notifyNum) const
175 : {
176 0 : *notifyNum = this->notifyNum_;
177 0 : return HCCL_SUCCESS;
178 : }
179 :
180 0 : HcclResult AicpuTsP2pChannel::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
181 : {
182 0 : return memTransport_->GetRemoteMems(memNum, remoteMem, memInfos);
183 : }
184 :
185 0 : ChannelStatus AicpuTsP2pChannel::GetStatus()
186 : {
187 0 : ChannelStatus out = Channel::TransportStatusToChannelStatus(memTransport_->GetStatus());
188 0 : return out;
189 : }
190 :
191 0 : HcclResult AicpuTsP2pChannel::SetModuleDataName(Hccl::ModuleData& module, const std::string& name)
192 : {
193 0 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
194 0 : if (ret != 0) {
195 0 : HCCL_ERROR("[SetModuleDataName] strcpy_s name %s failed", name.c_str());
196 0 : return HCCL_E_INTERNAL;
197 : }
198 :
199 0 : return HCCL_SUCCESS;
200 : }
201 :
202 0 : HcclResult AicpuTsP2pChannel::PackOpData(std::vector<char>& data)
203 : {
204 0 : std::vector<Hccl::ModuleData> dataVec;
205 0 : dataVec.resize(Hccl::AicpuResMgrType::__COUNT__);
206 :
207 0 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
208 0 : CHK_RET(SetModuleDataName(dataVec[resType], "P2PTransport"));
209 :
210 0 : std::vector<char> result;
211 0 : Hccl::BinaryStream binaryStream;
212 0 : binaryStream << memTransport_->GetUniqueIdV2();
213 :
214 0 : binaryStream.Dump(result);
215 :
216 0 : dataVec[resType].data = result;
217 :
218 : Hccl::AicpuResPackageHelper helper;
219 0 : data = helper.GetPackedData(dataVec);
220 :
221 0 : return HCCL_SUCCESS;
222 0 : }
223 :
224 0 : HcclResult AicpuTsP2pChannel::H2DResPack(std::vector<char>& buffer)
225 : {
226 0 : CHK_RET(PackOpData(buffer));
227 0 : HCCL_INFO(
228 : "[AicpuTsP2pChannel][%s] Pack Buffer data[%p], Pack Buffer size[%zu].", __func__, buffer.data(), buffer.size());
229 0 : return HCCL_SUCCESS;
230 : }
231 :
232 0 : HcclResult AicpuTsP2pChannel::Clean()
233 : {
234 0 : memTransport_.reset();
235 0 : return HCCL_SUCCESS;
236 : }
237 :
238 0 : HcclResult AicpuTsP2pChannel::Resume()
239 : {
240 0 : BuildSocket();
241 0 : BuildConnection();
242 0 : BuildP2pMemTransport();
243 0 : return HCCL_SUCCESS;
244 : }
245 :
246 : HcclResult
247 0 : AicpuTsP2pChannel::UpdateMemInfo([[maybe_unused]] HcommMemHandle* memHandles, [[maybe_unused]] uint32_t memHandleNum)
248 : {
249 0 : HCCL_WARNING("[AicpuTsP2pChannel][%s] P2PTransport does not support UpdateMemInfo.", __func__);
250 0 : return HCCL_SUCCESS;
251 : }
252 :
253 0 : HcclResult AicpuTsP2pChannel::NotifyRecord([[maybe_unused]] const uint32_t remoteNotifyIdx)
254 : {
255 0 : HCCL_INFO("[AicpuTsP2pChannel::%s] not supported yet.", __func__);
256 0 : return HCCL_E_NOT_SUPPORT;
257 : }
258 :
259 : HcclResult
260 0 : AicpuTsP2pChannel::NotifyWait([[maybe_unused]] const uint32_t localNotifyIdx, [[maybe_unused]] const uint32_t timeout)
261 : {
262 0 : HCCL_INFO("[AicpuTsP2pChannel::%s] not supported yet.", __func__);
263 0 : return HCCL_E_NOT_SUPPORT;
264 : }
265 :
266 0 : HcclResult AicpuTsP2pChannel::WriteWithNotify(
267 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] const uint64_t len,
268 : [[maybe_unused]] uint32_t remoteNotifyIdx)
269 : {
270 0 : HCCL_INFO("[AicpuTsP2pChannel::%s] not supported yet.", __func__);
271 0 : return HCCL_E_NOT_SUPPORT;
272 : }
273 :
274 : HcclResult
275 0 : AicpuTsP2pChannel::Write([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
276 : {
277 0 : HCCL_INFO("[AicpuTsP2pChannel::%s] not supported yet.", __func__);
278 0 : return HCCL_E_NOT_SUPPORT;
279 : }
280 :
281 : HcclResult
282 0 : AicpuTsP2pChannel::Read([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
283 : {
284 0 : HCCL_INFO("[AicpuTsP2pChannel::%s] not supported yet.", __func__);
285 0 : return HCCL_E_NOT_SUPPORT;
286 : }
287 :
288 0 : HcclResult AicpuTsP2pChannel::ChannelFence()
289 : {
290 0 : HCCL_INFO("[AicpuTsP2pChannel::%s] not supported yet.", __func__);
291 0 : return HCCL_E_NOT_SUPPORT;
292 : }
293 :
294 : } // namespace hcomm
|
