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