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_channel_ctx_pool.h"
12 :
13 : #include "ccu_device_pub.h"
14 : #include "orion_adpt_utils.h"
15 :
16 : namespace hcomm {
17 :
18 : constexpr uint32_t CCU_DEFAULT_REQUEST_SQ_SIZE = 128;
19 : constexpr uint32_t CCU_DEFAULT_REQUEST_CHANNEL_NUM = 1;
20 : constexpr uint32_t CCU_DEFAULT_REQUEST_JETTY_NUM = 0; // 申请数量为0时，由平台层决定提供数量
21 :
22 29 : CcuChannelCtxPool::CcuChannelCtxPool(int32_t devLogicId) : devLogicId_(devLogicId) {}
23 :
24 29 : CcuChannelCtxPool::~CcuChannelCtxPool()
25 : {
26 : // 对象析构时清空多个map，batchMap_中元素的jettys清空触发ccuJetty析构释放
27 29 : (void)ReleaseConfirmedChannelRes();
28 29 : }
29 :
30 15 : HcclResult CcuChannelCtxPool::ResourceBatch::Init(const std::vector<CcuChannelInfo>& channelInfos)
31 : {
32 15 : const uint32_t channelNum = channelInfos.size();
33 15 : channelIdKeys.reserve(channelNum);
34 15 : availableChannelIdKeys.reserve(channelNum);
35 44 : for (const auto& channelInfo : channelInfos) {
36 29 : const auto dieId = channelInfo.dieId;
37 29 : const auto channelId = channelInfo.channelId;
38 29 : channelIdKeys.emplace_back(dieId, channelId);
39 29 : availableChannelIdKeys.emplace_back(dieId, channelId);
40 :
41 58 : for (const auto& jettyInfo : channelInfo.jettyInfos) {
42 29 : const auto taJettyId = jettyInfo.taJettyId;
43 29 : const auto jettyIdKey = std::make_pair(dieId, taJettyId);
44 29 : if (jettys.find(jettyIdKey) != jettys.end()) {
45 14 : continue;
46 : }
47 :
48 15 : std::unique_ptr<CcuJetty> ccuJetty;
49 15 : CHK_RET(CcuCreateJetty(key, jettyInfo, ccuJetty));
50 :
51 15 : jettys[jettyIdKey] = std::move(ccuJetty);
52 15 : }
53 : }
54 :
55 15 : return HcclResult::HCCL_SUCCESS;
56 : }
57 :
58 15 : HcclResult CcuChannelCtxPool::PrepareCreate(const std::vector<Hccl::LinkData>& links, uint32_t sqSize)
59 : {
60 15 : std::lock_guard<std::mutex> lock(mtx_);
61 :
62 15 : CHK_PRT_RET(
63 : links.empty(),
64 : HCCL_INFO("[CcuChannelCtxPool][%s] passed, links is empty, devLogicId[%d].", __func__, devLogicId_),
65 : HcclResult::HCCL_SUCCESS);
66 :
67 30 : for (const auto& link : links) {
68 15 : auto it = allocatedChannelIdMap_.find(link);
69 15 : if (it != allocatedChannelIdMap_.end()) {
70 0 : HCCL_INFO(
71 : "[CcuChannelCtxPool][%s] passed, link[%s] is already allocated, "
72 : "devLogicId[%d].",
73 : __func__, link.Describe().c_str(), devLogicId_);
74 0 : continue;
75 : }
76 :
77 15 : const auto& locAddr = link.GetLocalAddr();
78 15 : ResourceBatch* batchPtr = nullptr;
79 15 : auto ret = GetAvailableBatch(locAddr, batchPtr, sqSize);
80 15 : CHK_PRT_RET(
81 : ret == HcclResult::HCCL_E_UNAVAIL,
82 : HCCL_WARNING(
83 : "[CcuChannelCtxPool][%s] failed to alloc ccu channels, ccu resources "
84 : "are unavailable, locAddr[%s], devLogicId[%d], sqSize[%u].",
85 : __func__, locAddr.Describe().c_str(), devLogicId_, sqSize),
86 : ret);
87 15 : CHK_RET(ret);
88 :
89 15 : ChannelIdKey channelIdKey = batchPtr->availableChannelIdKeys.back();
90 15 : batchPtr->availableChannelIdKeys.pop_back();
91 15 : allocatedChannelIdMap_[link] = channelIdKey;
92 15 : channelRemoteRankIdMap_[channelIdKey] = link.GetRemoteRankId();
93 15 : usedChannelCntMap_[channelIdKey.first] += 1;
94 :
95 15 : HCCL_INFO(
96 : "[CcuChannelCtxPool][%s] allocated new channelId[%u] of die[%u] to link[%s], "
97 : "devLogicId[%d], sqSize[%u].",
98 : __func__, channelIdKey.second, channelIdKey.first, link.Describe().c_str(), devLogicId_, sqSize);
99 : }
100 :
101 15 : isReleased_ = false;
102 15 : return HcclResult::HCCL_SUCCESS;
103 15 : }
104 :
105 : // 当前以locAddr为粒度调用，根据locAddr可以找到已申请的批次，如果资源充足则复用，不足则按新批次申请资源
106 15 : HcclResult CcuChannelCtxPool::GetAvailableBatch(const BatchKey& batchKey, ResourceBatch*& batchPtr, uint32_t sqSize)
107 : {
108 : // 当前以locAddr作为batchKey，不同本端不能复用资源
109 15 : if (FindAvailableBatch(batchKey, batchPtr)) {
110 0 : return HcclResult::HCCL_SUCCESS;
111 : }
112 : // 已有的资源不足，需要新增资源，获取的channel数量可能超过申请数量
113 15 : CommAddr commAddr{};
114 15 : CHK_RET(IpAddressToCommAddr(batchKey, commAddr));
115 : // 使用传入的sqSize，如果为0xFFFFFFFF则使用默认值
116 15 : uint32_t actualSqSize = (sqSize != 0xFFFFFFFF) ? sqSize : CCU_DEFAULT_REQUEST_SQ_SIZE;
117 : const CcuChannelPara channelPara{
118 15 : commAddr, CCU_DEFAULT_REQUEST_CHANNEL_NUM, CCU_DEFAULT_REQUEST_JETTY_NUM, actualSqSize};
119 15 : std::vector<CcuChannelInfo> channelInfos;
120 15 : auto ret = CcuAllocChannels(devLogicId_, channelPara, channelInfos);
121 15 : CHK_PRT_RET(
122 : ret == HcclResult::HCCL_E_UNAVAIL,
123 : HCCL_WARNING(
124 : "[CcuChannelCtxPool][%s] failed to alloc ccu channels, ccu resources "
125 : "are unavailable, locAddr[%s] devLogicId[%d].",
126 : __func__, batchKey.Describe().c_str(), devLogicId_),
127 : ret);
128 15 : CHK_RET(ret);
129 : // 如果新增资源保存失败，手动释放避免泄露
130 15 : ret = CreateAndSaveNewBatch(batchKey, channelInfos, batchPtr);
131 15 : if (ret != HcclResult::HCCL_SUCCESS) {
132 0 : HCCL_ERROR(
133 : "[CcuChannelCtxPool][%s] failed, try to release temp ccu resources, locAddr[%s], "
134 : "devLogicId[%d], .",
135 : __func__, batchKey.Describe().c_str(), devLogicId_);
136 0 : for (const auto& channelInfo : channelInfos) {
137 0 : const auto dieId = channelInfo.dieId;
138 0 : const auto channelId = channelInfo.channelId;
139 0 : CHK_RET(CcuReleaseChannel(devLogicId_, dieId, channelId));
140 : }
141 0 : return ret;
142 : }
143 15 : return HcclResult::HCCL_SUCCESS;
144 15 : }
145 :
146 15 : HcclResult CcuChannelCtxPool::CreateAndSaveNewBatch(
147 : const BatchKey& batchKey, const std::vector<CcuChannelInfo> channelInfos, ResourceBatch*& batchPtr)
148 : {
149 : // todo: 需要检查资源管理是否存在泄露可能
150 15 : auto& batches = batchMap_[batchKey];
151 15 : std::unique_ptr<ResourceBatch> newBatch{nullptr};
152 15 : newBatch.reset(new (std::nothrow) ResourceBatch(batchKey));
153 15 : CHK_PTR_NULL(newBatch);
154 15 : CHK_RET(newBatch->Init(channelInfos));
155 44 : for (const auto& channelInfo : channelInfos) {
156 29 : const auto dieId = channelInfo.dieId;
157 29 : const auto channelIdKey = std::make_pair(dieId, channelInfo.channelId);
158 :
159 29 : std::vector<CcuJetty*> jettys;
160 58 : for (const auto& jettyInfo : channelInfo.jettyInfos) {
161 29 : const auto jettyIdKey = std::make_pair(dieId, jettyInfo.taJettyId);
162 29 : jettys.emplace_back(newBatch->jettys[jettyIdKey].get());
163 : }
164 :
165 29 : channelJettyInfoMap_.emplace(channelIdKey, std::make_pair(channelInfo, jettys));
166 29 : }
167 :
168 15 : batches.push_back(std::move(newBatch));
169 15 : ResourceBatch* rawBatch = batches.back().get();
170 44 : for (const auto& channelInfo : channelInfos) {
171 29 : channelToBatch_[std::make_pair(channelInfo.dieId, channelInfo.channelId)] = rawBatch;
172 : }
173 15 : batchPtr = rawBatch;
174 15 : return HcclResult::HCCL_SUCCESS;
175 15 : }
176 :
177 15 : bool CcuChannelCtxPool::FindAvailableBatch(const BatchKey& batchKey, ResourceBatch*& batchPtr) const
178 : {
179 15 : auto it = batchMap_.find(batchKey);
180 15 : if (it == batchMap_.end()) {
181 15 : return false;
182 : }
183 :
184 0 : auto& batches = it->second;
185 : // 从后往前遍历：越晚创建的 batch 越可能留有可复用槽位（新申请通常分配自尾部），
186 : // 优先命中可减少扫描；中间 batch 释放出的槽位同样可被后续创建复用
187 0 : for (auto batchIter = batches.rbegin(); batchIter != batches.rend(); ++batchIter) {
188 0 : if (*batchIter != nullptr && !(*batchIter)->availableChannelIdKeys.empty()) {
189 0 : batchPtr = batchIter->get();
190 0 : return true;
191 : }
192 : }
193 0 : return false;
194 : }
195 :
196 : HcclResult
197 15 : CcuChannelCtxPool::GetChannelCtx(const Hccl::LinkData& link, CcuChannelCtxPool::CcuChannelCtx& channelCtx) const
198 : {
199 15 : std::lock_guard<std::mutex> lock(mtx_);
200 :
201 15 : const auto& it = allocatedChannelIdMap_.find(link);
202 15 : CHK_PRT_RET(
203 : it == allocatedChannelIdMap_.end(),
204 : HCCL_ERROR(
205 : "[CcuChannelCtxPool][%s] failed to find allocated channelId of link[%s], ", "devLogicId[%d].", __func__,
206 : link.Describe().c_str(), devLogicId_),
207 : HcclResult::HCCL_E_NOT_FOUND);
208 : // 内部维护数据保证channelJettyInfoMap_记录的资源存在
209 15 : channelCtx = channelJettyInfoMap_.at(it->second);
210 15 : return HcclResult::HCCL_SUCCESS;
211 15 : }
212 :
213 29 : HcclResult CcuChannelCtxPool::ReleaseConfirmedChannelRes()
214 : {
215 : // 析构路径唯一入口，内部持锁保护 map 遍历与设备层归还
216 29 : std::lock_guard<std::mutex> lock(mtx_);
217 :
218 29 : for (const auto& infoEntry : channelJettyInfoMap_) {
219 0 : const auto& channelIdKey = infoEntry.first;
220 0 : const auto dieId = channelIdKey.first;
221 0 : const auto channelId = channelIdKey.second;
222 0 : CHK_RET(CcuReleaseChannel(devLogicId_, dieId, channelId));
223 : }
224 29 : channelJettyInfoMap_.clear();
225 29 : channelToBatch_.clear();
226 29 : isReleased_ = true;
227 29 : return HcclResult::HCCL_SUCCESS;
228 29 : }
229 :
230 0 : HcclResult CcuChannelCtxPool::GetCcuChannelCtxById(const std::pair<uint8_t, uint32_t>& key, CcuChannelCtx& ctx)
231 : {
232 0 : std::lock_guard<std::mutex> lock(mtx_);
233 :
234 0 : auto it = channelJettyInfoMap_.find(key);
235 0 : if (it == channelJettyInfoMap_.end()) {
236 0 : HCCL_ERROR("[%s]fail, key[%u, %u] not found", __func__, key.first, key.second);
237 0 : return HCCL_E_NOT_FOUND;
238 : }
239 0 : ctx = it->second;
240 0 : return HCCL_SUCCESS;
241 0 : }
242 :
243 15 : HcclResult CcuChannelCtxPool::ReleaseChannel(const Hccl::LinkData& link)
244 : {
245 15 : std::lock_guard<std::mutex> lock(mtx_);
246 :
247 15 : auto it = allocatedChannelIdMap_.find(link);
248 15 : if (it == allocatedChannelIdMap_.end()) {
249 : // 未分配或已释放的 link 直接返回成功：msg-only(资源不足)等未实际分配资源的
250 : // channel 销毁路径属正常场景，静默返回即可
251 0 : HCCL_DEBUG("[CcuChannelCtxPool][%s] link not allocated, devLogicId[%d], skip release.", __func__, devLogicId_);
252 0 : return HcclResult::HCCL_SUCCESS;
253 : }
254 15 : const auto channelIdKey = it->second;
255 15 : allocatedChannelIdMap_.erase(link);
256 15 : channelRemoteRankIdMap_.erase(channelIdKey);
257 15 : auto cntIt = usedChannelCntMap_.find(channelIdKey.first);
258 15 : if (cntIt != usedChannelCntMap_.end() && cntIt->second > 0) {
259 15 : cntIt->second -= 1;
260 15 : if (cntIt->second == 0) {
261 15 : usedChannelCntMap_.erase(cntIt);
262 : }
263 : }
264 :
265 15 : ResourceBatch* batch = FindBatchByChannelId(channelIdKey);
266 15 : if (batch == nullptr) {
267 0 : HCCL_ERROR(
268 : "[CcuChannelCtxPool][%s] failed to find batch of channelId[%u] die[%u], "
269 : "devLogicId[%d].",
270 : __func__, channelIdKey.second, channelIdKey.first, devLogicId_);
271 0 : return HcclResult::HCCL_E_INTERNAL;
272 : }
273 : // 槽位压回可复用列表：V2 组内其他 channel 仍活跃时，设备层占用保持不变，
274 : // 后续创建可直接复用该槽位（连接流程会重新配置 channel 表）。
275 15 : batch->availableChannelIdKeys.push_back(channelIdKey);
276 : // 整组无活跃 channel 时，锁内整组归还设备层并销毁 batch
277 15 : HcclResult releaseRet = ReleaseBatchIfIdle(batch);
278 15 : if (releaseRet != HcclResult::HCCL_SUCCESS) {
279 : // 设备层归还失败：该槽位不能被后续创建复用（设备层资源并未真正归还），
280 : // 从可复用列表回退，资源留待通信域销毁时 ReleaseConfirmedChannelRes 兜底重试
281 0 : if (!batch->availableChannelIdKeys.empty() && batch->availableChannelIdKeys.back() == channelIdKey) {
282 0 : batch->availableChannelIdKeys.pop_back();
283 : }
284 : }
285 15 : return releaseRet;
286 15 : }
287 :
288 15 : CcuChannelCtxPool::ResourceBatch* CcuChannelCtxPool::FindBatchByChannelId(const ChannelIdKey& key) const
289 : {
290 15 : auto it = channelToBatch_.find(key);
291 15 : return (it == channelToBatch_.end()) ? nullptr : it->second;
292 : }
293 :
294 15 : HcclResult CcuChannelCtxPool::ReleaseBatchIfIdle(CcuChannelCtxPool::ResourceBatch* batch)
295 : {
296 15 : if (batch->availableChannelIdKeys.size() != batch->channelIdKeys.size()) {
297 : // 组内仍有活跃 channel，保留 batch 供槽位复用
298 0 : return HcclResult::HCCL_SUCCESS;
299 : }
300 : // 整组无活跃 channel：逐 channel 归还设备层（V2 useCnt 递减至 0），全部成功才销毁 batch；
301 : // 任一失败则保留 batch（channelJettyInfoMap_/channelToBatch_ 条目仍在，host 对象不被销毁），
302 : // 返回错误供上层感知；通信域销毁时的 ReleaseConfirmedChannelRes 会再次尝试整体归还。
303 44 : for (const auto& channelIdKey : batch->channelIdKeys) {
304 29 : auto ret = CcuReleaseChannel(devLogicId_, channelIdKey.first, channelIdKey.second);
305 29 : if (ret != HcclResult::HCCL_SUCCESS) {
306 0 : HCCL_ERROR(
307 : "[CcuChannelCtxPool][%s] failed to release channel[die%u, id%u] to device, "
308 : "ret[%d], devLogicId[%d], keep batch for retry.",
309 : __func__, channelIdKey.first, channelIdKey.second, ret, devLogicId_);
310 0 : return ret;
311 : }
312 : }
313 15 : RemoveBatch(batch);
314 15 : return HcclResult::HCCL_SUCCESS;
315 : }
316 :
317 15 : void CcuChannelCtxPool::RemoveBatch(CcuChannelCtxPool::ResourceBatch* batch)
318 : {
319 15 : auto it = batchMap_.find(batch->key);
320 15 : if (it == batchMap_.end()) {
321 0 : return;
322 : }
323 15 : auto& batches = it->second;
324 15 : for (auto bIt = batches.begin(); bIt != batches.end(); ++bIt) {
325 15 : if (bIt->get() != batch) {
326 0 : continue;
327 : }
328 44 : for (const auto& channelIdKey : batch->channelIdKeys) {
329 29 : channelJettyInfoMap_.erase(channelIdKey);
330 29 : channelToBatch_.erase(channelIdKey);
331 : }
332 : // 锁内销毁 batch：~ResourceBatch → ~CcuJetty → Clean → RaCtxQpDestroy
333 15 : batches.erase(bIt);
334 15 : break;
335 : }
336 15 : if (batches.empty()) {
337 15 : batchMap_.erase(it);
338 : }
339 : }
340 : } // namespace hcomm
|
