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 "ccu_transport_manager.h"
12 :
13 : #include <chrono>
14 :
15 : #include "coll_operator.h"
16 : #include "exception_util.h"
17 : #include "socket_manager.h"
18 : #include "ccu_communicator.h"
19 : #include "communicator_impl.h"
20 : #include "timeout_exception.h"
21 : #include "internal_exception.h"
22 : #include "coll_service_device_mode.h"
23 : #include "adapter_error_manager_pub.h"
24 :
25 : namespace Hccl {
26 :
27 279 : CcuTransportMgr::CcuTransportMgr(const CommunicatorImpl &comm, const int32_t devLogicId)
28 279 : : comm(&comm), devLogicId_(devLogicId)
29 : {
30 279 : }
31 :
32 279 : CcuTransportMgr::~CcuTransportMgr()
33 : {
34 279 : if (!isDestroyed) {
35 278 : DECTOR_TRY_CATCH("CcuTransportMgr", Destroy());
36 : }
37 279 : }
38 :
39 6 : CcuTransport *CcuTransportMgr::Get(const LinkData &link)
40 : {
41 6 : auto linkIter = ccuLink2TransportMap.find(link);
42 6 : if (linkIter != ccuLink2TransportMap.end()) {
43 4 : return linkIter->second.get();
44 : }
45 6 : HCCL_WARNING("[CcuTransportMgr::%s] CcuTransport does not existed, "
46 : "errNo[0x%016llx], localRank[%d], remoteRank[%d]", __func__,
47 : HCCL_ERROR_CODE(HcclResult::HCCL_E_PTR), link.GetLocalRankId(), link.GetRemoteRankId());
48 :
49 2 : return nullptr;
50 : }
51 :
52 4 : set<CcuTransport*> CcuTransportMgr::Get(RankId rank)
53 : {
54 4 : auto rankIter = ccuRank2TransportsMap.find(rank);
55 4 : if (rankIter != ccuRank2TransportsMap.end()) {
56 2 : return rankIter->second;
57 : }
58 6 : HCCL_WARNING("[CcuTransportMgr::%s] CcuTransport does not existed, "
59 : "errNo[0x%016llx], remoteRank[%d]", __func__,
60 : HCCL_ERROR_CODE(HcclResult::HCCL_E_PTR), rank);
61 2 : return set<CcuTransport*>();
62 : }
63 :
64 13 : HcclResult CcuTransportMgr::PrepareCreate(const LinkData &link, CcuTransport *&transport)
65 : {
66 13 : auto linkIter = ccuLink2TransportMap.find(link);
67 13 : if (linkIter != ccuLink2TransportMap.end()) {
68 1 : transport = linkIter->second.get();
69 1 : return HcclResult::HCCL_SUCCESS;
70 : }
71 :
72 12 : auto ret = CreateTransportByLink(link, transport);
73 12 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
74 3 : HCCL_WARNING("[CcuTransportMgr::%s]Fail to create CcuTransport. "
75 : "The above error log can be ignores.", __func__);
76 1 : comm->PrintChannelInfoCallback();
77 : }
78 :
79 12 : return ret;
80 : }
81 :
82 12 : static HcclResult CheckIfLinkProtocolSupport(const LinkData &link)
83 : {
84 12 : const auto linkProtocol = link.GetLinkProtocol();
85 12 : if (link.GetLinkProtocol() != LinkProtocol::UB_CTP && linkProtocol != LinkProtocol::UB_TP) {
86 0 : HCCL_ERROR("[CcuTransportMgr][%s] %s is not supported now, only ub_ctp/ub_tp can be created, "
87 : "please check, link[%s].", __func__, linkProtocol.Describe().c_str(),
88 : link.Describe().c_str());
89 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
90 : }
91 :
92 12 : return HcclResult::HCCL_SUCCESS;
93 : }
94 :
95 12 : HcclResult CcuTransportMgr::CreateTransportByLink(const LinkData &link, CcuTransport *&transport)
96 : {
97 36 : HCCL_INFO("[CcuTransportMgr][%s] begain", __func__);
98 12 : CHECK_NULLPTR(comm, "[CcuTransportMgr::CreateTransportByLink] comm is nullptr!");
99 12 : CHK_RET(CheckIfLinkProtocolSupport(link));
100 :
101 12 : std::string socketTag = comm->GetEstablishLinkSocketTag();
102 12 : SocketConfig socketConfig(link.GetRemoteRankId(), link, socketTag);
103 12 : Socket *socket = comm->GetSocketManager().GetConnectedSocket(socketConfig);
104 12 : if (socket == nullptr) {
105 0 : HCCL_WARNING("[CcuTransportMgr::%s] Fail to get socket via link %s, ",
106 : __func__, link.Describe().c_str());
107 0 : return HcclResult::HCCL_E_INTERNAL;
108 : }
109 :
110 24 : CcuJettyMgr *ccuJettyMgr = dynamic_cast<CollServiceDeviceMode *>(comm->GetCollService())
111 24 : ->GetCcuInsPreprocessor()->GetCcuComm()->GetCcuJettyMgr();
112 12 : const auto channelJettys = ccuJettyMgr->GetChannelJettys(link);
113 12 : const CcuChannelInfo &channelInfo = channelJettys.first;
114 12 : const std::vector<CcuJetty *> &ccuJettys = channelJettys.second;
115 :
116 12 : const auto &locAddr = link.GetLocalAddr();
117 12 : const auto &rmtAddr = link.GetRemoteAddr();
118 12 : CcuTransport::CcuConnectionType type = link.GetLinkProtocol() == LinkProtocol::UB_CTP ?
119 12 : CcuTransport::CcuConnectionType::UBC_CTP : CcuTransport::CcuConnectionType::UBC_TP;
120 12 : CcuTransport::CcuConnectionInfo connectionInfo{type, locAddr, rmtAddr, channelInfo, ccuJettys};
121 :
122 12 : std::shared_ptr<LocalUbRmaBuffer> locCclRmaBuffer;
123 12 : if (comm->GetCclBuffer() == nullptr) {
124 0 : HCCL_ERROR("dataBuf[type=SCRATCH] is nullptr");
125 0 : return HcclResult::HCCL_E_INTERNAL;
126 : }
127 36 : HCCL_INFO("[CcuTransportMgr][%s] comm cclBuf[%s]", __func__, comm->GetCclBuffer()->Describe().c_str());
128 12 : locCclRmaBuffer = make_shared<LocalUbRmaBuffer>(comm->GetCclBuffer());
129 :
130 36 : HCCL_INFO("[CcuTransportMgr::CreateTransportByLink] locCclRmaBuffer[%s]", locCclRmaBuffer->Describe().c_str());
131 : const CcuTransport::CclBufferInfo locCclBufInfo {
132 12 : locCclRmaBuffer->GetBuf()->GetAddr(),
133 12 : static_cast<uint32_t>(locCclRmaBuffer->GetBuf()->GetSize()),
134 : locCclRmaBuffer->GetTokenId(),
135 : locCclRmaBuffer->GetTokenValue()
136 24 : };
137 :
138 : // 当前不支持创建非UBC协议的链路
139 12 : std::unique_ptr<CcuTransport> transportPtr = nullptr;
140 12 : auto ret = CcuCreateTransport(socket, connectionInfo, locCclBufInfo, transportPtr);
141 12 : if (ret == HcclResult::HCCL_E_UNAVAIL) {
142 3 : HCCL_WARNING("[CcuTransportMgr][%s] failed, some ccu resources are unavaialble, "
143 : "locAddr[%s] rmtAddr[%s].", __func__, locAddr.Describe().c_str(), rmtAddr.Describe().c_str());
144 1 : return ret;
145 : }
146 11 : CHK_RET(ret);
147 :
148 11 : tempTransport.emplace_back(link);
149 11 : ccuLink2TransportMap[link] = std::move(transportPtr);
150 11 : const auto &rawTransportPtr = ccuLink2TransportMap[link].get();
151 11 : ccuRank2TransportsMap[link.GetRemoteRankId()].insert(rawTransportPtr);
152 :
153 11 : transport = rawTransportPtr;
154 33 : HCCL_INFO("[CcuTransportMgr][%s] end", __func__);
155 11 : return HcclResult::HCCL_SUCCESS;
156 12 : }
157 :
158 9 : void CcuTransportMgr::WaitTransportsReady(vector<std::pair<CcuTransport*, LinkData>> &transports) const
159 : {
160 9 : auto timeout = std::chrono::seconds(EnvConfig::GetInstance().GetSocketConfig().GetLinkTimeOut());
161 9 : HcclUs startTime = std::chrono::steady_clock::now();
162 13 : while (!transports.empty()) {
163 11 : for (auto transIter = transports.begin(); transIter != transports.end();) {
164 7 : auto status = (*transIter).first->GetStatus();
165 7 : if (status == CcuTransport::TransStatus::CONNECT_FAILED) {
166 2 : THROW<InternalException>("Invalid status occurs when creating transport connection %s!",
167 6 : (*transIter).first->Describe().c_str());
168 : }
169 :
170 5 : if (status == CcuTransport::TransStatus::SOCKET_TIMEOUT) {
171 1 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}),
172 : std::vector<std::string>({"CcuTransport wait SOCKET_TIMEOUT."}));
173 2 : THROW<TimeoutException>("[CcuTransportMgr][%s] [CcuTransport]%s [LinkData]%s socket timeout, "
174 3 : "commId[%s], please check.", __func__, (*transIter).first->Describe().c_str(),
175 3 : (*transIter).second.Describe().c_str(), comm->GetId().c_str());
176 : }
177 :
178 4 : if (status != CcuTransport::TransStatus::READY) {
179 0 : ++transIter;
180 0 : continue;
181 : }
182 4 : transIter = transports.erase(transIter);
183 : }
184 :
185 4 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
186 0 : string timeoutMsg = StringFormat("CcuTransportMgr::WaitTransportReady timeout, commId[%s]", comm->GetId().c_str());
187 0 : HCCL_ERROR(timeoutMsg.c_str());
188 0 : DumpNotReadyTransports(transports);
189 : // 上报EI0006
190 0 : RPT_INPUT_ERR(true, "EI0006", std::vector<std::string>({"reason"}),
191 : std::vector<std::string>({"CcuTransportMgr wait transports ready timeout."}));
192 0 : THROW<InternalException>(timeoutMsg);
193 0 : }
194 : }
195 6 : }
196 :
197 1 : void CcuTransportMgr::DumpNotReadyTransports(vector<std::pair<CcuTransport*, LinkData>> &transports) const
198 : {
199 3 : HCCL_ERROR("Dump ccu timeout transport info, transport size[%u]", transports.size());
200 2 : for (auto transIter = transports.begin(); transIter != transports.end(); ++transIter) {
201 1 : string allStr = (*transIter).first->Describe();
202 1 : size_t pos = allStr.find("Socket");
203 1 : if(pos != string::npos) {
204 3 : HCCL_ERROR("CcuTransport[%s]", allStr.substr(0,pos).c_str());
205 1 : allStr=allStr.substr(pos);
206 : }
207 3 : HCCL_ERROR("CcuTransport[%s]", allStr.c_str());
208 3 : HCCL_ERROR("LinkData[%s]", (*transIter).second.Describe().c_str());
209 1 : }
210 1 : }
211 :
212 9 : void CcuTransportMgr::TransportsConnect()
213 : {
214 9 : vector<std::pair<CcuTransport*, LinkData>> transLinkPairs = GetUnConfirmedTrans();
215 9 : auto op = comm->GetCurrentCollOperator();
216 9 : auto accelerator = comm->GetOpExecuteConfig().accState;
217 27 : HCCL_INFO("[CcuTransportMgr::TransportsConnect] accelerator[%s]", accelerator.Describe().c_str());
218 :
219 16 : for (auto &pair : transLinkPairs) {
220 7 : auto transport = pair.first;
221 7 : transport->SetLocalOpAcceState(accelerator);
222 7 : transport->SetHandshakeMsg(op->GetUniqueId());
223 :
224 21 : HCCL_INFO("[CcuTransportMgr::%s] transport=[%s]", __func__, transport->Describe().c_str());
225 21 : HCCL_INFO("[CcuTransportMgr::%s] links=[%s]", __func__, pair.second.Describe().c_str());
226 21 : HCCL_INFO("[CcuTransportMgr::%s] opInfo=[%s]", __func__, CollOpToString(*op).c_str());
227 : }
228 :
229 9 : WaitTransportsReady(transLinkPairs);
230 :
231 18 : HCCL_INFO("[CcuTransportMgr::%s] transports connect end.", __func__);
232 9 : }
233 :
234 5 : void CcuTransportMgr::Confirm()
235 : {
236 5 : TransportsConnect();
237 5 : tempTransport.clear();
238 5 : }
239 :
240 7 : vector<std::pair<CcuTransport *, LinkData>> CcuTransportMgr::GetUnConfirmedTrans()
241 : {
242 7 : if (tempTransport.size() == 0) {
243 6 : HCCL_WARNING("[CcuTransportMgr::%s] UnConfirmedTrans does not exist, please check.", __func__);
244 2 : return vector<std::pair<CcuTransport *, LinkData>>();
245 : }
246 :
247 5 : vector<std::pair<CcuTransport *, LinkData>> unConfirmedTrans;
248 10 : for (const auto &linkData : tempTransport) {
249 5 : auto iterLink = ccuLink2TransportMap.find(linkData);
250 5 : if (iterLink == ccuLink2TransportMap.end()) {
251 0 : THROW<InternalException>("[CcuTransportMgr::%s]Link can't find, linkData[%s]", __func__,
252 0 : linkData.Describe().c_str());
253 : }
254 5 : unConfirmedTrans.emplace_back(std::make_pair(iterLink->second.get(), linkData));
255 : }
256 5 : return unConfirmedTrans;
257 5 : }
258 :
259 287 : void CcuTransportMgr::Clean()
260 : {
261 287 : BatchDeleteJettyInfo batchDeleteJettyInfo;
262 : // 获取所有transport的unimportJetty和deleteJetty
263 307 : for (auto &linkTransPair : ccuLink2TransportMap) {
264 20 : if (linkTransPair.second == nullptr) {
265 1 : continue;
266 : }
267 19 : auto partDeleteInfo = linkTransPair.second->GetDeleteJettyInfo();
268 43 : for (auto& jettyInfo : partDeleteInfo) {
269 24 : if (jettyInfo.localJetty != 0) {
270 14 : batchDeleteJettyInfo.deleteJettyList[jettyInfo.rdmaHandle].insert(jettyInfo.localJetty);
271 : }
272 : }
273 19 : auto partUnimportInfo = linkTransPair.second->GetUnimportJettyInfo();
274 33 : for (auto& jettyInfo : partUnimportInfo) {
275 14 : if (jettyInfo.remoteJetty != 0) {
276 14 : batchDeleteJettyInfo.unimportJettyList[jettyInfo.rdmaHandle].insert(jettyInfo.remoteJetty);
277 : }
278 : }
279 19 : }
280 :
281 : // 循环unimport
282 294 : for (auto& tmp : batchDeleteJettyInfo.unimportJettyList) {
283 7 : auto& unimportJettys = tmp.second;
284 21 : for (auto& unimportJetty : unimportJettys) {
285 14 : HrtRaUbUnimportJetty(tmp.first, unimportJetty);
286 : }
287 : }
288 :
289 : // 批量销毁jetty
290 287 : std::vector<JettyHandle> failJettyHandles;
291 294 : for (auto& tmp : batchDeleteJettyInfo.deleteJettyList) {
292 7 : auto& rdmaHandle = tmp.first;
293 7 : auto& delJettys = tmp.second;
294 7 : auto ret = HrtRaCtxQpDestoryBatch(rdmaHandle, delJettys, failJettyHandles);
295 9 : for (u64 failJetty : failJettyHandles) {
296 6 : HCCL_ERROR("[%s]delete jetty[%llu] fail", __func__, failJetty);
297 : }
298 7 : if (ret == HCCL_E_INTERNAL || ret == HCCL_E_TIMEOUT) {
299 9 : HCCL_ERROR("[%s]HrtRaCtxQpDestoryBatch finish, ret[%u], rdmaHandle[%p], originalJettyCount[%u], undeleteJettyCount[%u]",
300 : __func__, ret, rdmaHandle, delJettys.size(), failJettyHandles.size());
301 3 : continue;
302 3 : } else {
303 12 : HCCL_INFO("[%s]HrtRaCtxQpDestoryBatch finish, ret[%u], rdmaHandle[%p], originalJettyCount[%u], undeleteJettyCount[%u]",
304 : __func__, ret, rdmaHandle, delJettys.size(), failJettyHandles.size());
305 : }
306 4 : failJettyHandles.clear();
307 : }
308 :
309 : // 清理transport
310 307 : for (auto &linkTransPair : ccuLink2TransportMap) {
311 20 : if (linkTransPair.second == nullptr) {
312 1 : continue;
313 : }
314 19 : if (linkTransPair.second->Clean() != HCCL_SUCCESS) {
315 0 : THROW<CcuApiException>("[CcuTransportMgr::%s]CcuTransport clean failed.", __func__);
316 : }
317 : }
318 287 : }
319 :
320 3 : void CcuTransportMgr::Resume()
321 : {
322 5 : for (auto iter = ccuLink2TransportMap.begin(); iter != ccuLink2TransportMap.end(); iter++) {
323 2 : tempTransport.push_back(iter->first);
324 : }
325 3 : }
326 :
327 4 : void CcuTransportMgr::Fallback()
328 : {
329 : // 遍历TempTransport所有link，分别在ccuLink2TransportMap和ccuRank2TransportsMap删除对应的Transport
330 4 : for (const auto &linkId : tempTransport) {
331 0 : auto iterLink = ccuLink2TransportMap.find(linkId);
332 : // 在ccuRank2TransportsMap中要删除的Transport
333 0 : auto prepareDelTransport = std::move(iterLink->second);
334 0 : ccuLink2TransportMap.erase(iterLink);
335 :
336 0 : auto iterRank = ccuRank2TransportsMap.find(linkId.GetRemoteRankId());
337 0 : auto iterRankSet = std::move(iterRank->second).find(prepareDelTransport.get());
338 0 : if (iterRankSet != std::move(iterRank->second).end()) {
339 0 : iterRank->second.erase(iterRankSet);
340 0 : if (iterRank->second.size() == 0) {
341 0 : ccuRank2TransportsMap.erase(iterRank);
342 : }
343 : }
344 0 : }
345 :
346 4 : tempTransport.clear();
347 4 : }
348 :
349 279 : void CcuTransportMgr::Destroy()
350 : {
351 279 : isDestroyed = true;
352 279 : Clean();
353 279 : ccuLink2TransportMap.clear();
354 279 : ccuRank2TransportsMap.clear();
355 279 : }
356 :
357 2 : void CcuTransportMgr::RecoverTransportsConnect()
358 : {
359 2 : vector<std::pair<CcuTransport *, LinkData>> transLinkPairs = GetUnConfirmedTrans();
360 2 : auto accelerator = comm->GetOpExecuteConfig().accState;
361 6 : HCCL_INFO("[CcuTransportMgr::TransportsConnect] accelerator[%s]", accelerator.Describe().c_str());
362 4 : for (auto &pair : transLinkPairs) {
363 2 : auto transport = pair.first;
364 :
365 2 : u32 crcValue{0};
366 6 : HCCL_INFO("[RecoverMemTransport]commptr=%p", comm);
367 :
368 2 : if (comm->IsWorldGroup()) {
369 : // 判断是否在框内
370 0 : if (comm->GetNeighboorRanks().find(pair.second.GetRemoteRankId()) != comm->GetNeighboorRanks().end()) {
371 : // 在框内使用带LocalID的CRC值
372 0 : crcValue = comm->GetRanktableCrc(true);
373 : } else {
374 : // 不在框内使用不带LocalID的CRC值
375 0 : crcValue = comm->GetRanktableCrc(false);
376 : }
377 : }
378 :
379 : // 握手消息定义，包括 通信算子数目，rankTable CRC，通信步骤字段
380 2 : CollOperator op{};
381 2 : op.opTag = std::to_string(comm->GetCollOpIndex()) + "_" + std::to_string(crcValue) + "_" + std::to_string(comm->GetStep());
382 2 : transport->SetLocalOpAcceState(accelerator);
383 2 : transport->SetHandshakeMsg(op.GetUniqueId());
384 6 : HCCL_INFO("[CcuTransportMgr::%s] transport=[%s]", __func__, transport->Describe().c_str());
385 6 : HCCL_INFO("[CcuTransportMgr::%s] links=[%s]", __func__, pair.second.Describe().c_str());
386 2 : }
387 :
388 2 : WaitTransportsRecoverReady(transLinkPairs);
389 :
390 3 : HCCL_INFO("[CcuTransportMgr::%s] transports connect end.", __func__);
391 2 : }
392 :
393 3 : void CcuTransportMgr::RecoverConfirm()
394 : {
395 3 : RecoverTransportsConnect();
396 2 : tempTransport.clear();
397 2 : }
398 :
399 2 : void CcuTransportMgr::WaitTransportsRecoverReady(vector<std::pair<CcuTransport*, LinkData>> &transports) const
400 : {
401 2 : constexpr u32 waitTransportReadyTimeoutMs = 10 * 1000; // 待修改，定义最大等待10秒
402 :
403 2 : auto timeout = std::chrono::milliseconds(waitTransportReadyTimeoutMs);
404 2 : HcclUs startTime = std::chrono::steady_clock::now();
405 3 : while (!transports.empty()) {
406 3 : for (auto transIter = transports.begin(); transIter != transports.end();) {
407 2 : auto status = (*transIter).first->GetStatus();
408 2 : if (status == CcuTransport::TransStatus::CONNECT_FAILED) {
409 1 : THROW<InternalException>("Invalid status occurs when creating transport connection %s!",
410 3 : (*transIter).first->Describe().c_str());
411 : }
412 :
413 1 : if (status != CcuTransport::TransStatus::READY) {
414 0 : ++transIter;
415 0 : continue;
416 : }
417 1 : transIter = transports.erase(transIter);
418 : }
419 :
420 1 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
421 0 : THROW<InternalException>("WaitTransportReady timeout, commId[%s]", comm->GetId().c_str());
422 : }
423 : }
424 1 : }
425 :
426 : } // namespace Hccl
|
