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