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_jetty_mgr.h"
12 :
13 : #include <unordered_set>
14 :
15 : #include "ccu_dev_mgr.h"
16 : #include "internal_exception.h"
17 : #include "invalid_params_exception.h"
18 :
19 : namespace Hccl {
20 :
21 : constexpr uint32_t CCU_DEFAULT_REQUEST_SQ_SIZE = 128;
22 : constexpr uint32_t CCU_CLOS_REQUEST_SQ_SIZE = 16; // CLOS场景的默认SQ大小
23 : constexpr uint32_t CCU_DEFAULT_REQUEST_CHANNEL_NUM = 1;
24 : constexpr uint32_t CCU_DEFAULT_REQUEST_JETTY_NUM = 0; // 申请数量为0时，由平台层决定提供数量
25 :
26 270 : CcuJettyMgr::CcuJettyMgr(int32_t devLogicId) : devLogicId_(devLogicId) {}
27 :
28 270 : CcuJettyMgr::~CcuJettyMgr()
29 : {
30 : // 对象析构时清空多个map，batchMap_中元素的jettys清空触发ccuJetty析构释放
31 270 : DECTOR_TRY_CATCH("CcuJettyMgr", ReleaseConfirmedChannelRes());
32 270 : }
33 :
34 14 : HcclResult CcuJettyMgr::PrepareCreate(const std::vector<LinkData>& links)
35 : {
36 17 : CHK_PRT_RET(
37 : links.empty(), HCCL_INFO("[CcuJettyMgr][%s] passed, links is empty, devLogicId[%d].", __func__, devLogicId_),
38 : HcclResult::HCCL_SUCCESS);
39 :
40 277 : for (const auto& link : links) {
41 267 : auto it = allocatedChannelIdMap_.find(link);
42 267 : if (it != allocatedChannelIdMap_.end()) {
43 144 : HCCL_INFO(
44 : "[CcuJettyMgr][%s] passed, link[%s] is already allocated, "
45 : "devLogicId[%d].",
46 : __func__, link.Describe().c_str(), devLogicId_);
47 48 : continue;
48 48 : }
49 :
50 219 : const auto& locAddr = link.GetLocalAddr();
51 : // 根据 LinkData 的 fullmesh 字段选择不同的 sqSize
52 219 : uint32_t sqSize = link.GetFullmesh() ? CCU_DEFAULT_REQUEST_SQ_SIZE : CCU_CLOS_REQUEST_SQ_SIZE;
53 219 : ResourceBatch* batchPtr = nullptr;
54 : // 这里仅使用locAddr作为batchKey，是因为不会存在相同locAddr但不同资源的场景（不同locAddr必然不同资源），如果后续有更复杂的场景再调整key设计
55 219 : auto ret = GetAvailableBatch(locAddr, batchPtr, sqSize);
56 224 : CHK_PRT_RET(
57 : ret == HcclResult::HCCL_E_UNAVAIL,
58 : HCCL_WARNING(
59 : "[CcuJettyMgr][%s] failed to alloc ccu channels, ccu resources "
60 : "are unavaialble, locAddr[%s], devLogicId[%d].",
61 : __func__, locAddr.Describe().c_str(), devLogicId_),
62 : ret);
63 224 : CHK_RET(ret);
64 :
65 216 : ChannelIdKey channelIdKey = batchPtr->availableChannelIdKeys.back();
66 216 : batchPtr->availableChannelIdKeys.pop_back();
67 216 : unconfirmedRecord_.allocations.emplace_back(Allocation{link, channelIdKey, batchPtr});
68 216 : allocatedChannelIdMap_[link] = channelIdKey;
69 216 : channelRemoteRankIdMap_[channelIdKey] = link.GetRemoteRankId();
70 216 : channelIpAddressMap_[channelIdKey] = {link.GetLocalAddr(), link.GetRemoteAddr()};
71 :
72 648 : HCCL_INFO(
73 : "[CcuJettyMgr][%s] allocated new channelId[%u] of die[%u] to link[%s], "
74 : "devLogicId[%d].",
75 : __func__, channelIdKey.second, channelIdKey.first, link.Describe().c_str(), devLogicId_);
76 : }
77 :
78 10 : isReleased = false;
79 10 : return HcclResult::HCCL_SUCCESS;
80 : }
81 :
82 : // 当前以locAddr为粒度调用，根据locAddr可以找到已申请的批次，如果资源充足则复用，不足则按新批次申请资源
83 219 : HcclResult CcuJettyMgr::GetAvailableBatch(const BatchKey& batchKey, ResourceBatch*& batchPtr, uint32_t sqSize)
84 : {
85 : // 当前以locAddr作为batchKey，不同本端不能复用资源
86 219 : if (FindAvailableBatch(batchKey, batchPtr)) {
87 170 : return HcclResult::HCCL_SUCCESS;
88 : }
89 : // 已有的资源不足，需要新增资源，获取的channel数量可能超过申请数量
90 49 : const CcuChannelPara channelPara{batchKey, CCU_DEFAULT_REQUEST_CHANNEL_NUM, CCU_DEFAULT_REQUEST_JETTY_NUM, sqSize};
91 147 : HCCL_INFO(
92 : "[CcuJettyMgr][%s] try to alloc ccu channels with channelPara[channelNum=%u, jettyNum=%u, sqSize=%u], "
93 : "locAddr[%s], devLogicId[%d].",
94 : __func__, channelPara.channelNum, channelPara.jettyNum, channelPara.sqSize, batchKey.Describe().c_str(),
95 : devLogicId_);
96 49 : std::vector<CcuChannelInfo> channelInfos;
97 49 : auto ret = CcuAllocChannels(devLogicId_, channelPara, channelInfos);
98 : // 如果资源不足，平台层返回不可用错误，需要上层感知不可用进行降级处理
99 52 : CHK_PRT_RET(
100 : ret == HcclResult::HCCL_E_UNAVAIL,
101 : HCCL_WARNING(
102 : "[CcuJettyMgr][%s] failed to alloc ccu channels, ccu resources "
103 : "are unavaialble, locAddr[%s], sqSize[%u], devLogicId[%d].",
104 : __func__, batchKey.Describe().c_str(), sqSize, devLogicId_),
105 : ret);
106 : // 其他错误直接返回错误码
107 51 : CHK_PRT_RET(
108 : ret != HCCL_SUCCESS,
109 : HCCL_ERROR(
110 : "[CcuJettyMgr][%s] failed to alloc ccu channels, ccu resources "
111 : "are unavaialble, locAddr[%s], sqSize[%u], devLogicId[%d].",
112 : __func__, batchKey.Describe().c_str(), sqSize, devLogicId_),
113 : ret);
114 : // 如果新增资源保存失败，手动释放避免泄露
115 47 : ret = CreateAndSaveNewBatch(batchKey, channelInfos, batchPtr);
116 47 : if (ret != HcclResult::HCCL_SUCCESS) {
117 3 : HCCL_ERROR(
118 : "[CcuJettyMgr][%s] failed, try to release temp ccu resources, locAddr[%s], "
119 : "devLogicId[%d], .",
120 : __func__, batchKey.Describe().c_str(), devLogicId_);
121 9 : for (const auto& channelInfo : channelInfos) {
122 8 : const auto dieId = channelInfo.dieId;
123 8 : const auto channelId = channelInfo.channelId;
124 8 : CHK_RET(CcuReleaseChannel(devLogicId_, dieId, channelId));
125 : }
126 1 : return ret;
127 : }
128 46 : return HcclResult::HCCL_SUCCESS;
129 49 : }
130 :
131 46 : HcclResult CcuJettyMgr::CreateAndSaveNewBatch(
132 : const BatchKey& batchKey, const std::vector<CcuChannelInfo> channelInfos, ResourceBatch*& batchPtr)
133 : {
134 : // 该流程如果抛异常，可能资源还未记录，析构无法释放导致资源泄露，故捕获异常处理
135 1150 : TRY_CATCH_RETURN(auto& batches = batchMap_[batchKey];
136 : auto newBatch = std::make_unique<ResourceBatch>(batchKey, channelInfos);
137 : for (const auto& channelInfo
138 : : channelInfos) {
139 : const auto dieId = channelInfo.dieId;
140 : const auto channelIdKey = std::make_pair(dieId, channelInfo.channelId);
141 :
142 : std::vector<CcuJetty*> jettys;
143 : for (const auto& jettyInfo : channelInfo.jettyInfos) {
144 : const auto jettyIdKey = std::make_pair(dieId, jettyInfo.taJettyId);
145 : jettys.emplace_back(newBatch->jettys[jettyIdKey].get());
146 : }
147 :
148 : channelJettyInfoMap_.emplace(channelIdKey, std::make_pair(channelInfo, jettys));
149 : usedChannelCntMap_[dieId] += 1;
150 : }
151 :
152 : batches.push_back(std::move(newBatch));
153 : ResourceBatch* rawBatch = batches.back().get();
154 :
155 : unconfirmedRecord_.newBatchSet.insert(rawBatch); batchPtr = rawBatch;);
156 46 : return HcclResult::HCCL_SUCCESS;
157 : }
158 :
159 219 : bool CcuJettyMgr::FindAvailableBatch(const BatchKey& batchKey, ResourceBatch*& batchPtr) const
160 : {
161 219 : auto it = batchMap_.find(batchKey);
162 219 : if (it == batchMap_.end()) {
163 39 : return false;
164 : }
165 :
166 180 : auto& batches = it->second;
167 180 : if (batches.empty()) {
168 0 : return false;
169 : }
170 : // 当前分配逻辑只有最后一个batch可能还有空闲资源
171 180 : auto& lastBatch = batches.back();
172 180 : if (lastBatch->availableChannelIdKeys.empty()) {
173 10 : return false;
174 : }
175 :
176 170 : batchPtr = lastBatch.get();
177 170 : return true;
178 : }
179 :
180 276 : CcuJettyMgr::CcuChannelJettyInfo CcuJettyMgr::GetChannelJettys(const LinkData& link) const
181 : {
182 276 : const auto& it = allocatedChannelIdMap_.find(link);
183 276 : CHK_RET_THROW(
184 : InternalException,
185 : StringFormat(
186 : "[CcuJettyMgr][%s] failed to find allocated channelId of link[%s], ", "devLogicId[%d].", __func__,
187 : link.Describe().c_str(), devLogicId_),
188 : it == allocatedChannelIdMap_.end());
189 : // 内部维护数据保证channelJettyInfoMap_记录的资源存在
190 552 : return channelJettyInfoMap_.at(it->second);
191 : }
192 :
193 15 : void CcuJettyMgr::Confirm() { unconfirmedRecord_.Clear(); }
194 :
195 5 : void CcuJettyMgr::Fallback()
196 : {
197 5 : FallbackAndRemoveBatches();
198 5 : FallbackAllocatedChannelJettyInfo();
199 5 : Confirm();
200 5 : }
201 :
202 5 : void CcuJettyMgr::FallbackAndRemoveBatches()
203 : {
204 : using BatchSetKey = std::unordered_map<BatchKey, std::unordered_set<ResourceBatch*>>;
205 5 : BatchSetKey batchesToRemoveByKey;
206 21 : for (const auto& allocation : unconfirmedRecord_.allocations) {
207 16 : ResourceBatch* batchPtr = allocation.batchPtr;
208 16 : if (!batchPtr) {
209 0 : continue;
210 : }
211 :
212 16 : const auto& newBatchSet = unconfirmedRecord_.newBatchSet;
213 16 : bool isNewBatch = newBatchSet.find(batchPtr) != newBatchSet.end();
214 16 : if (!isNewBatch) { // 如果是之前申请的资源需要回退到待分配状态
215 0 : batchPtr->availableChannelIdKeys.push_back(allocation.channelIdKey);
216 0 : continue;
217 : }
218 :
219 16 : batchesToRemoveByKey[batchPtr->key].insert(batchPtr);
220 : }
221 : // 本轮新申请的资源需要调用接口释放，使用set去重
222 9 : for (const auto& keyEntry : batchesToRemoveByKey) {
223 4 : const auto& key = keyEntry.first;
224 4 : const auto& batchesToRemove = keyEntry.second;
225 :
226 8 : for (const auto& batchPtr : batchesToRemove) {
227 36 : for (const auto& channelIdKey : batchPtr->channelIdKeys) {
228 32 : const auto dieId = channelIdKey.first;
229 32 : const auto channelId = channelIdKey.second;
230 32 : CHK_RET_THROW(
231 : InternalException,
232 : StringFormat(
233 : "[CcuJettyMgr][%s] failed to release channelId[%u] of "
234 : "die[%u], devLogicId[%d].",
235 : __func__, channelId, dieId, devLogicId_),
236 : CcuReleaseChannel(devLogicId_, dieId, channelId));
237 :
238 32 : channelJettyInfoMap_.erase(channelIdKey); // 清理新增但未分配的资源信息
239 32 : usedChannelCntMap_[dieId] -= 1;
240 : }
241 : }
242 : // 从batchMap_中清理新增的batch
243 4 : auto& batches = batchMap_[key];
244 8 : batches.erase(
245 4 : std::remove_if(
246 : batches.begin(), batches.end(),
247 4 : [&batchesToRemove](const auto& batchPtr) {
248 4 : return batchesToRemove.find(batchPtr.get()) != batchesToRemove.end();
249 : }),
250 4 : batches.end());
251 :
252 4 : if (batches.empty()) {
253 4 : batchMap_.erase(key);
254 : }
255 : }
256 5 : }
257 :
258 5 : void CcuJettyMgr::FallbackAllocatedChannelJettyInfo()
259 : {
260 : // 清理非batch粒度记录的需回退的资源
261 21 : for (const auto& allocation : unconfirmedRecord_.allocations) {
262 16 : allocatedChannelIdMap_.erase(allocation.link);
263 16 : channelJettyInfoMap_.erase(allocation.channelIdKey);
264 16 : channelRemoteRankIdMap_.erase(allocation.channelIdKey);
265 16 : channelIpAddressMap_.erase(allocation.channelIdKey);
266 : }
267 5 : }
268 :
269 1 : void CcuJettyMgr::Clean()
270 : {
271 1 : ReleaseConfirmedChannelRes();
272 : // n秒快恢需要记录链路信息，故仅清空channel信息
273 17 : for (auto& linkEntry : allocatedChannelIdMap_) {
274 16 : linkEntry.second = {0, 0};
275 : }
276 :
277 1 : batchMap_.clear();
278 1 : channelJettyInfoMap_.clear();
279 1 : channelRemoteRankIdMap_.clear();
280 1 : channelIpAddressMap_.clear();
281 1 : usedChannelCntMap_.clear();
282 1 : Confirm();
283 1 : }
284 :
285 271 : void CcuJettyMgr::ReleaseConfirmedChannelRes()
286 : {
287 619 : for (const auto& infoEntry : channelJettyInfoMap_) {
288 348 : const auto& channelIdKey = infoEntry.first;
289 348 : const auto dieId = channelIdKey.first;
290 348 : const auto channelId = channelIdKey.second;
291 348 : CHK_RET_THROW(
292 : InternalException,
293 : StringFormat(
294 : "[CcuJettyMgr][%s] failed to release channelId[%u] of "
295 : "die[%u], devLogicId[%d].",
296 : __func__, channelId, dieId, devLogicId_),
297 : CcuReleaseChannel(devLogicId_, dieId, channelId));
298 : }
299 271 : isReleased = true;
300 271 : }
301 :
302 1 : void CcuJettyMgr::Resume()
303 : {
304 1 : if (!isReleased) {
305 0 : THROW<InternalException>(
306 : "[CcuJettyMgr][%s] failed, the ccu resources "
307 : "have not been released yet, devLogicId[%d].",
308 : __func__, devLogicId_);
309 : }
310 :
311 1 : std::vector<LinkData> links;
312 1 : links.reserve(allocatedChannelIdMap_.size());
313 17 : for (const auto& linkEntry : allocatedChannelIdMap_) {
314 16 : links.push_back(linkEntry.first);
315 : }
316 1 : allocatedChannelIdMap_.clear();
317 :
318 1 : CHK_RET_THROW(
319 : InternalException,
320 : StringFormat("[CcuJettyMgr][%s] failed to resume ccu jettys, devLogicId[%d].", __func__, devLogicId_),
321 : PrepareCreate(links));
322 1 : }
323 :
324 2 : uint32_t CcuJettyMgr::GetUsedChannelCount(const uint8_t dieId)
325 : {
326 2 : const auto& dieIter = usedChannelCntMap_.find(dieId);
327 2 : if (dieIter != usedChannelCntMap_.end()) {
328 1 : return dieIter->second;
329 : }
330 1 : return 0;
331 : }
332 :
333 1 : RankId CcuJettyMgr::GetRemoteRankIdByChannelId(const uint8_t dieId, const uint32_t channelId)
334 : {
335 1 : const auto& iter = channelRemoteRankIdMap_.find({dieId, channelId});
336 1 : if (iter == channelRemoteRankIdMap_.end()) {
337 0 : THROW<InvalidParamsException>(
338 : "[CcuJettyMgr][%s] failed to find remoteRankId by "
339 : "dieId[%u] channelId[%u], devLogicId[%d].",
340 : __func__, dieId, channelId, devLogicId_);
341 : }
342 :
343 2 : return iter->second;
344 : }
345 :
346 0 : std::pair<IpAddress, IpAddress> CcuJettyMgr::GetAddrPairByChannelId(const uint8_t dieId, const uint32_t channelId)
347 : {
348 0 : const auto& iter = channelIpAddressMap_.find({dieId, channelId});
349 0 : if (iter == channelIpAddressMap_.end()) {
350 0 : THROW<InvalidParamsException>(
351 : "[CcuJettyMgr][%s] failed to find addrPair by "
352 : "dieId[%u] channelId[%u], devLogicId[%d].",
353 : __func__, dieId, channelId, devLogicId_);
354 : }
355 :
356 0 : return iter->second;
357 : }
358 :
359 : } // namespace Hccl
|
