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