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