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 "rma_conn_manager.h"
12 : #include "p2p_connection.h"
13 : #include "communicator_impl.h"
14 : #include "rdma_handle_manager.h"
15 : #include "dev_rdma_connection.h"
16 : #include "dev_ub_connection.h"
17 : #include "null_ptr_exception.h"
18 : #include "exception_util.h"
19 : #include "socket_manager.h"
20 : #include "timeout_exception.h"
21 : namespace Hccl {
22 :
23 519 : RmaConnManager::RmaConnManager(const CommunicatorImpl &comm)
24 519 : : isDestroyed(false), comm(&comm)
25 : {
26 1557 : HCCL_INFO("AICPU: RmaConnManager init");
27 519 : }
28 :
29 1036 : RmaConnManager::~RmaConnManager()
30 : {
31 519 : if (!isDestroyed) {
32 519 : DECTOR_TRY_CATCH("RmaConnManager", Destroy());
33 : }
34 1036 : }
35 :
36 0 : unique_ptr<RmaConnection> RmaConnManager::CreateRdmaConn(Socket *socket, const std::string &tag,
37 : const LinkData &linkData) const
38 : {
39 0 : CHECK_NULLPTR(socket, "[RmaConnManager::CreateRdmaConn] socket is nullptr!");
40 0 : RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(
41 0 : comm->GetDevicePhyId(), linkData.GetLocalPort(), linkData.GetLinkProtocol());
42 :
43 0 : OpMode opMode = comm->GetCurrentCollOperator()->opMode;
44 0 : unique_ptr<DevRdmaConnection> rmaNetConn = make_unique<DevRdmaConnection>(socket, rdmaHandle, opMode);
45 0 : QpHandle qpHandle = rmaNetConn->GetHandle();
46 :
47 0 : auto buffer = comm->GetDataBufferManager().Get(tag, BufferType::SCRATCH);
48 0 : if (buffer == nullptr) {
49 0 : THROW<NullPtrException>(StringFormat("RmaConnManager::CreateRdmaConn ptr is null"));
50 : }
51 0 : RaMrInfo bufInfo{};
52 0 : bufInfo.addr = reinterpret_cast<void *>(buffer->GetAddr());
53 0 : bufInfo.size = buffer->GetSize();
54 0 : bufInfo.access = static_cast<u32>(RA_ACCESS_LOCAL_WRITE) | static_cast<u32>(RA_ACCESS_REMOTE_WRITE);
55 0 : HrtRaMrReg(qpHandle, bufInfo);
56 0 : return std::unique_ptr<RmaConnection>(rmaNetConn.release());
57 0 : }
58 :
59 0 : unique_ptr<RmaConnection> RmaConnManager::CreateUbConn(Socket *socket, const std::string &tag,
60 : const LinkData &linkData, const HrtUbJfcMode jfcMode)
61 : {
62 0 : RdmaHandle rdmaHandle = RdmaHandleManager::GetInstance().Get(comm->GetDevicePhyId(), linkData.GetLocalPort(), linkData.GetLinkProtocol());
63 0 : OpMode opMode = comm->GetCurrentCollOperator()->opMode;
64 0 : HCCL_INFO("[RmaConnManager][%s]opMode[%d],linkData[%s],devicePhyId[%u], tag[%s]",
65 : __func__, static_cast<int32_t>(opMode), linkData.Describe().c_str(),
66 : comm->GetDevicePhyId(), tag.c_str());
67 :
68 0 : unique_ptr<DevUbConnection> ubConn = nullptr;
69 0 : locAddr = linkData.GetLocalAddr();
70 0 : rmtAddr = linkData.GetRemoteAddr();
71 0 : IpAddress locIpv4Addr = locAddr;
72 0 : IpAddress rmtIpv4Addr = rmtAddr;
73 0 : HCCL_INFO("[RmaConnManager][%s] LinkProtocol[%s], locAddr[%s], rmtAddr[%s]",
74 : __func__, linkData.GetLinkProtocol().Describe().c_str(), locAddr.Describe().c_str(), rmtAddr.Describe().c_str());
75 0 : if (linkData.GetLinkProtocol() == LinkProtocol::UBOE || linkData.GetLinkProtocol() == LinkProtocol::UBG) {
76 : // socket建链状态ok，并交换数据
77 0 : WaitUboeSocketReady(socket, linkData);
78 : }
79 :
80 0 : bool devUsed = comm->GetOpAiCpuTSFeatureFlag();
81 0 : if (linkData.GetLinkProtocol() == LinkProtocol::UB_TP) {
82 0 : ubConn = make_unique<DevUbTpConnection>(rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode);
83 0 : } else if (linkData.GetLinkProtocol() == LinkProtocol::UBOE) {
84 0 : ubConn = make_unique<DevUbUboeConnection>(rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode, locIpv4Addr, rmtIpv4Addr);
85 0 : } else if (linkData.GetLinkProtocol() == LinkProtocol::UBG) {
86 0 : ubConn = make_unique<DevUbUbgConnection>(rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode, locAddr, rmtAddr);
87 : } else {
88 0 : ubConn = make_unique<DevUbCtpConnection>(rdmaHandle, locAddr, rmtAddr, opMode, devUsed, jfcMode);
89 : }
90 0 : return std::unique_ptr<RmaConnection>(ubConn.release());
91 0 : }
92 :
93 0 : RmaConnection *RmaConnManager::Create(const std::string &tag, const LinkData &linkData, const HrtUbJfcMode jfcMode)
94 : {
95 0 : HCCL_INFO("Create tag = [%s], remoteRank[%d] LinkData[%s] ", tag.c_str(), linkData.GetRemoteRankId(),
96 : linkData.Describe().c_str());
97 0 : RmaConnection *rmaConnPtr = Get(tag, linkData);
98 0 : if (rmaConnPtr != nullptr) {
99 0 : HCCL_INFO("has inited");
100 0 : return rmaConnPtr;
101 : }
102 :
103 0 : std::string socketTag = comm->GetEstablishLinkSocketTag();
104 0 : SocketConfig socketConfig(linkData.GetRemoteRankId(), linkData, socketTag);
105 0 : Socket *socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
106 0 : HCCL_INFO("socketTag = [%s]", socketTag.c_str());
107 0 : HCCL_INFO("[RmaConnManager::%s] linkData Type[%s]", __func__, linkData.GetType().Describe().c_str());
108 0 : std::unique_ptr<RmaConnection> rmaConn = nullptr;
109 0 : if (linkData.GetType() == PortDeploymentType::P2P) {
110 0 : rmaConn = make_unique<P2PConnection>(socket, tag);
111 0 : } else if (linkData.GetType() == PortDeploymentType::DEV_NET) {
112 0 : auto linkProtocol = linkData.GetLinkProtocol();
113 0 : HCCL_INFO("[RmaConnManager::%s] linkData linkProtocol[%s]", __func__, linkProtocol.Describe().c_str());
114 0 : if (linkProtocol == LinkProtocol::ROCE) {
115 0 : rmaConn = CreateRdmaConn(socket, tag, linkData);
116 0 : } else if (linkProtocol == LinkProtocol::UB_TP || linkProtocol == LinkProtocol::UB_CTP ||
117 0 : linkProtocol == LinkProtocol::UBOE || linkProtocol == LinkProtocol::UBG) {
118 0 : rmaConn = CreateUbConn(socket, tag, linkData, jfcMode);
119 : }
120 : }
121 :
122 0 : if (rmaConn == nullptr) {
123 0 : auto msg = StringFormat("Fail to create RmaConnection via link %s", linkData.Describe().c_str());
124 0 : THROW<NullPtrException>(msg.c_str());
125 0 : }
126 :
127 0 : rmaConn->Connect();
128 0 : rmaConnectionMap[tag][linkData] = std::move(rmaConn);
129 :
130 0 : return rmaConnectionMap[tag][linkData].get();
131 0 : }
132 :
133 0 : void RmaConnManager::RecreateAllConns()
134 : {
135 0 : for (const auto &connPair : rmaConnectionMap) {
136 0 : const string &tag = connPair.first;
137 0 : for (const auto &linkDataConnPair : connPair.second) {
138 0 : const LinkData &linkData = linkDataConnPair.first;
139 0 : if (linkDataConnPair.second != nullptr) {
140 0 : rmaConnectionMap[tag][linkData] = nullptr;
141 0 : Create(tag, linkData);
142 : }
143 : }
144 : }
145 0 : }
146 :
147 0 : RmaConnection *RmaConnManager::Get(const std::string &tag, const LinkData &linkData)
148 : {
149 0 : auto tagIter = rmaConnectionMap.find(tag);
150 0 : if (tagIter != rmaConnectionMap.end()) {
151 0 : auto linkDataIter = tagIter->second.find(linkData);
152 0 : if (linkDataIter != tagIter->second.end()) {
153 0 : return linkDataIter->second.get();
154 : }
155 : }
156 0 : HCCL_WARNING("WARNING: RmaConnection not existed, "
157 : "errNo[0x%016llx], localRank[%d], remoteRank[%d], tag[%s]",
158 : HCCL_ERROR_CODE(HcclResult::HCCL_E_PTR), comm->GetMyRank(), linkData.GetRemoteRankId(), tag.c_str());
159 :
160 0 : return nullptr;
161 : }
162 :
163 8 : std::vector<RmaConnection *> RmaConnManager::GetOpTagConns(const std::string &tag) const
164 : {
165 8 : std::vector<RmaConnection *> rmaConnList;
166 8 : auto opTagIter = rmaConnectionMap.find(tag);
167 8 : if (opTagIter != rmaConnectionMap.end()) {
168 4 : for (auto &linkDataConn : opTagIter->second) {
169 2 : rmaConnList.emplace_back(linkDataConn.second.get());
170 : }
171 2 : return rmaConnList;
172 : }
173 18 : HCCL_WARNING("WARNING: RmaConnection not existed, "
174 : "errNo[0x%016llx], localRank[%d], tag[%s]",
175 : HCCL_ERROR_CODE(HcclResult::HCCL_E_PTR), comm->GetMyRank(), tag.c_str());
176 6 : return rmaConnList;
177 0 : }
178 :
179 0 : void RmaConnManager::Release(const std::string &tag, const LinkData &linkData)
180 : {
181 0 : auto tagIter = rmaConnectionMap.find(tag);
182 0 : if (tagIter != rmaConnectionMap.end()) {
183 0 : auto linkDataIter = tagIter->second.find(linkData);
184 0 : if (linkDataIter != tagIter->second.end()) {
185 0 : tagIter->second.erase(linkDataIter);
186 : }
187 : }
188 0 : }
189 :
190 519 : void RmaConnManager::GetDeleteJettys(BatchDeleteJettyInfo &batchDeleteJettyInfo)
191 : {
192 : // 获取要删除的连接
193 519 : DevUbConnection* ubConn = nullptr;
194 521 : for (auto &connPair : rmaConnectionMap) {
195 4 : for (auto &linkDataConnPair : connPair.second) {
196 2 : if (linkDataConnPair.second != nullptr) {
197 2 : ubConn = dynamic_cast<DevUbConnection*>(linkDataConnPair.second.get());
198 2 : if (ubConn == nullptr) {
199 0 : continue;
200 : }
201 :
202 2 : const auto& rdmaHandle = ubConn->GetRdmaHandle();
203 2 : auto& remoteJettyHandle = ubConn->GetRemoteJettyHandle();
204 2 : if (rdmaHandle && remoteJettyHandle != 0) {
205 0 : batchDeleteJettyInfo.unimportJettyList[rdmaHandle].insert(remoteJettyHandle);
206 0 : remoteJettyHandle = 0;
207 : }
208 2 : ubConn->ReleaseTp();
209 2 : auto& jettyHandle = ubConn->GetJettyHandle();
210 2 : if (jettyHandle != 0) {
211 0 : batchDeleteJettyInfo.deleteJettyList[rdmaHandle].insert(jettyHandle);
212 0 : jettyHandle = 0;
213 : }
214 2 : linkDataConnPair.second = nullptr;
215 : }
216 : }
217 : }
218 519 : }
219 :
220 519 : void RmaConnManager::BatchDeleteJettys()
221 : {
222 519 : BatchDeleteJettyInfo batchDeleteJettyInfo;
223 519 : GetDeleteJettys(batchDeleteJettyInfo);
224 519 : for(auto& unimportJettys : batchDeleteJettyInfo.unimportJettyList) {
225 0 : for(auto& unimportJetty : unimportJettys.second) {
226 0 : HrtRaUbUnimportJetty(unimportJettys.first, unimportJetty);
227 : }
228 : }
229 :
230 519 : std::vector<JettyHandle> failJettyHandles;
231 519 : for(const auto& deleteJettys : batchDeleteJettyInfo.deleteJettyList) {
232 0 : auto ret = HrtRaCtxQpDestoryBatch(deleteJettys.first, deleteJettys.second, failJettyHandles);
233 0 : for (u64 failJetty : failJettyHandles) {
234 0 : HCCL_ERROR("[%s]delete jetty[%llu] fail", __func__, failJetty);
235 : }
236 0 : if (ret == HCCL_E_INTERNAL || ret == HCCL_E_TIMEOUT) {
237 0 : HCCL_ERROR("[%s]HrtRaCtxQpDestoryBatch finish, ret[%u], rdmaHandle[%p], originalJettyCount[%u], undeleteJettyCount[%u]",
238 : __func__, ret, deleteJettys.first, deleteJettys.second.size(), failJettyHandles.size());
239 0 : continue;
240 0 : } else {
241 0 : HCCL_INFO("[%s]HrtRaCtxQpDestoryBatch finish, ret[%u], rdmaHandle[%p], originalJettyCount[%u], undeleteJettyCount[%u]",
242 : __func__, ret, deleteJettys.first, deleteJettys.second.size(), failJettyHandles.size());
243 : }
244 0 : failJettyHandles.clear();
245 : }
246 519 : }
247 :
248 519 : void RmaConnManager::Destroy()
249 : {
250 519 : isDestroyed = true;
251 519 : BatchDeleteJettys();
252 519 : rmaConnectionMap.clear();
253 519 : }
254 :
255 0 : void RmaConnManager::Clear()
256 : {
257 0 : BatchDeleteJettys();
258 0 : rmaConnectionMap.clear();
259 0 : }
260 :
261 0 : std::vector<RmaConnection *> RmaConnManager::GetAllConns() const
262 : {
263 0 : std::vector<RmaConnection *> rmaConnList;
264 0 : for (const auto &connPair : rmaConnectionMap) {
265 0 : for (const auto &linkDataConnPair : connPair.second) {
266 0 : if (linkDataConnPair.second != nullptr) {
267 0 : rmaConnList.push_back(linkDataConnPair.second.get());
268 : }
269 : }
270 : }
271 0 : return rmaConnList;
272 0 : }
273 :
274 : const std::vector<BufferType> BUF_TYPES = {BufferType::SCRATCH, BufferType::INPUT, BufferType::OUTPUT};
275 :
276 0 : void RmaConnManager::BindRemoteRmaBuffers()
277 : {
278 0 : for (const auto &connPair : rmaConnectionMap) {
279 0 : const string &tag = connPair.first;
280 0 : for (const auto &linkDataConnPair : connPair.second) {
281 0 : const LinkData &linkData = linkDataConnPair.first;
282 0 : for (auto &bufType : BUF_TYPES) {
283 : RemoteRmaBuffer *remoteRmaBuf
284 0 : = comm->GetRemoteRmaBufManager().GetRemoteRmaBuffer(tag, linkData, bufType);
285 0 : if (remoteRmaBuf != nullptr) {
286 0 : rmaConnectionMap[tag][linkData]->Bind(remoteRmaBuf, bufType);
287 : }
288 : }
289 : }
290 : }
291 0 : }
292 :
293 0 : void RmaConnManager::BatchCreate(vector<LinkData> &links)
294 : {
295 0 : HCCL_INFO("[NsRecovery][Resume]RmaConnManager::BatchCreate, before Create, rmaConnectionMap size[%u]",
296 : rmaConnectionMap.size());
297 0 : const string &tag = comm->GetId();
298 0 : for (const auto &linkData : links) {
299 0 : if (rmaConnectionMap[tag][linkData] == nullptr) {
300 0 : Create(tag, linkData);
301 : } else {
302 0 : HCCL_WARNING("[NsRecovery][Resume]RmaConnManager::BatchCreate, connection has existed, will not recreate, "
303 : "linkData[%s]", linkData.Describe().c_str());
304 : }
305 : }
306 0 : HCCL_INFO("[NsRecovery][Resume]RmaConnManager::BatchCreate, after Create, rmaConnectionMap size[%u], "
307 : "rmaConnectionMap[comm->GetId()] size[%u]",
308 : rmaConnectionMap.size(), rmaConnectionMap[tag].size());
309 0 : }
310 :
311 0 : bool RmaConnManager::IsSocketReady(Socket *socket, const LinkData &linkData)
312 : {
313 0 : if (socket == nullptr) {
314 0 : MACRO_THROW(InternalException, StringFormat("%s socket is nullptr, please check", linkData.Describe().c_str()));
315 : }
316 :
317 0 : SocketStatus socketStatus = socket->GetAsyncStatus();
318 0 : if (socketStatus == SocketStatus::OK) {
319 0 : uboeStatus = UboeStatus::SOCKET_OK;
320 0 : return true;
321 0 : } else if (socketStatus == SocketStatus::TIMEOUT) {
322 0 : uboeStatus = UboeStatus::SOCKET_TIMEOUT;
323 0 : return false;
324 : }
325 :
326 0 : return false;
327 : }
328 :
329 0 : UboeStatus RmaConnManager::GetUboeSocketStatus(Socket *socket, const LinkData &linkData)
330 : {
331 0 : if (uboeStatus == UboeStatus::READY) {
332 0 : return uboeStatus;
333 0 : } else if (uboeStatus == UboeStatus::INIT) {
334 0 : ubStatus = UbStatus::INIT;
335 : }
336 :
337 0 : if (!IsSocketReady(socket, linkData)) {
338 0 : return uboeStatus;
339 : }
340 :
341 0 : switch (ubStatus) {
342 0 : case UbStatus::INIT:
343 0 : ubStatus = UbStatus::SOCKET_OK;
344 0 : uboeStatus = UboeStatus::SOCKET_OK;
345 0 : break;
346 0 : case UbStatus::SOCKET_OK:
347 0 : ubStatus = UbStatus::SEND_DATA;
348 0 : SendExchangeData(socket, linkData);
349 0 : break;
350 0 : case UbStatus::SEND_DATA:
351 0 : RecvExchangeData(socket, linkData);
352 0 : ubStatus = UbStatus::RECV_DATA;
353 0 : break;
354 0 : case UbStatus::RECV_DATA:
355 0 : RecvDataProcess(linkData);
356 0 : ubStatus = UbStatus::RECV_FIN;
357 0 : uboeStatus = UboeStatus::READY;
358 0 : break;
359 0 : default:
360 0 : break;
361 : }
362 0 : return uboeStatus;
363 : }
364 :
365 0 : void RmaConnManager::WaitUboeSocketReady(Socket *socket, const LinkData &linkData)
366 : {
367 0 : HCCL_INFO("[RmaConnManager][%s] begain", __func__);
368 0 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
369 0 : HcclUs startTime = std::chrono::steady_clock::now();
370 : while (true) {
371 0 : auto status = GetUboeSocketStatus(socket, linkData);
372 0 : if (status == UboeStatus::READY) {
373 0 : break;
374 : }
375 0 : if (status == UboeStatus::SOCKET_TIMEOUT) {
376 0 : MACRO_THROW(TimeoutException,
377 : StringFormat("[RmaConnManager][%s] %s socket timeout, commId[%s], please check",
378 : __func__, linkData.Describe().c_str(),
379 : comm->GetId().c_str()));
380 : }
381 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
382 0 : string timeoutMsg = StringFormat("WaitUboeSocketReady timeout, commId[%s].", comm->GetId().c_str());
383 0 : HCCL_ERROR(timeoutMsg.c_str());
384 0 : THROW<InternalException>(timeoutMsg);
385 0 : }
386 0 : }
387 0 : HCCL_INFO("[RmaConnManager][%s] end", __func__);
388 0 : }
389 :
390 0 : void RmaConnManager::Ipv4Pack()
391 : {
392 0 : IpAddress locEidAddr;
393 0 : RdmaHandleManager::GetInstance().GetEidByIpv4Addr(locAddr, locEidAddr);
394 0 : locAddr = locEidAddr;
395 0 : HCCL_INFO("[RmaConnManager::%s] locAddr[%s]", __func__, locAddr.Describe().c_str());
396 0 : sendData = locAddr.GetUniqueId();
397 0 : }
398 :
399 0 : void RmaConnManager::Ipv4UnPack(BinaryStream& binaryStream)
400 : {
401 0 : IpAddress rmtEidAddr(binaryStream);
402 0 : rmtAddr = rmtEidAddr;
403 0 : HCCL_INFO("[RmaConnManager::%s] rmtAddr[%s]", __func__, rmtAddr.Describe().c_str());
404 0 : }
405 :
406 0 : void RmaConnManager::SendExchangeData(Socket *socket, const LinkData &linkData)
407 : {
408 0 : Ipv4Pack();
409 0 : socket->SendAsync(sendData.data(), sendData.size());
410 0 : exchangeDataSize = sendData.size();
411 :
412 0 : HCCL_INFO("send data %s, size=%llu", linkData.Describe().c_str(), exchangeDataSize);
413 0 : }
414 :
415 0 : void RmaConnManager::RecvExchangeData(Socket *socket, const LinkData &linkData)
416 : {
417 0 : recvData.resize(exchangeDataSize);
418 0 : socket->RecvAsync(reinterpret_cast<u8 *>(recvData.data()), recvData.size());
419 :
420 0 : HCCL_INFO("recv data %s, size=%llu", linkData.Describe().c_str(), recvData.size());
421 0 : }
422 :
423 0 : void RmaConnManager::RecvDataProcess(const LinkData &linkData)
424 : {
425 0 : HCCL_INFO("RecvDataProcess: link=%s, size=%llu, exchangeDataSize=%u", linkData.Describe().c_str(), recvData.size(),
426 : exchangeDataSize);
427 0 : BinaryStream binaryStream(recvData);
428 0 : Ipv4UnPack(binaryStream);
429 0 : }
430 :
431 : } // namespace Hccl
|
