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 "subscribe_manager.h"
12 : #include "common/bqs_log.h"
13 : #include "driver/ascend_hal.h"
14 : #include "statistic_manager.h"
15 : #include "bqs_util.h"
16 :
17 : namespace bqs {
18 0 : SubscribeManager &SubscribeManager::GetInstance()
19 : {
20 0 : static SubscribeManager instance;
21 0 : return instance;
22 : }
23 :
24 74 : void SubscribeManager::InitSubscribeManager(const uint32_t deviceId, const uint32_t enqueGroupId,
25 : const uint32_t f2nfGroupId, const uint32_t dstDeviceId)
26 : {
27 74 : deviceId_ = deviceId;
28 74 : enqueGroupId_ = enqueGroupId;
29 74 : f2nfGroupId_ = f2nfGroupId;
30 74 : dstDeviceId_ = dstDeviceId;
31 74 : extendDriverInterface_ = (GetRunContext() == RunContext::HOST) || GlobalCfg::GetInstance().GetNumaFlag();
32 74 : subscribeQueuesMaps_.clear();
33 74 : fullToNotFullQueuesSets_.clear();
34 74 : }
35 :
36 135 : BqsStatus SubscribeManager::Subscribe(uint32_t queueId)
37 : {
38 135 : const auto findRet = subscribeQueuesMaps_.find(queueId);
39 135 : if (findRet != subscribeQueuesMaps_.end()) {
40 4 : BQS_LOG_INFO("queue[%u] devId[%u] is subscribed, no need to subscribe.", queueId, deviceId_);
41 4 : return BQS_STATUS_OK;
42 : }
43 :
44 131 : const auto ret = EnhancedSubscribe(queueId, QUEUE_ENQUE_EVENT);
45 131 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_QUEUE_RE_SUBSCRIBED)) {
46 3 : BQS_LOG_ERROR("halQueueSubscribe queue[%u] on device[%u] in group[%u] failed, ret=%d.",
47 : queueId, deviceId_, enqueGroupId_, static_cast<int32_t>(ret));
48 3 : return BQS_STATUS_DRIVER_ERROR;
49 : }
50 :
51 128 : if ((ret == DRV_ERROR_QUEUE_RE_SUBSCRIBED) && (Resubscribe(queueId) != BQS_STATUS_OK)) {
52 2 : BQS_LOG_ERROR("ReSubscribeBuffQueue queue[%u] on device[%u] in group[%u] failed.",
53 : queueId, deviceId_, enqueGroupId_);
54 2 : return BQS_STATUS_DRIVER_ERROR;
55 : }
56 :
57 126 : (void)subscribeQueuesMaps_.insert(std::make_pair(queueId, true));
58 126 : BQS_LOG_INFO("halQueueSubscribe queue[%u], deviceId[%u] in group[%u] success.", queueId, deviceId_, enqueGroupId_);
59 126 : return BQS_STATUS_OK;
60 : }
61 :
62 6 : BqsStatus SubscribeManager::UpdateSubscribe(const uint32_t queueId)
63 : {
64 6 : const auto findRet = subscribeQueuesMaps_.find(queueId);
65 6 : if (findRet == subscribeQueuesMaps_.end()) {
66 2 : BQS_LOG_INFO("queue[%u] devId[%u] is not subscribed, no need to update subscribe.", queueId, deviceId_);
67 2 : return BQS_STATUS_OK;
68 : }
69 :
70 4 : const auto status = EnhancedSubscribe(queueId, QUEUE_ENQUE_EVENT);
71 4 : if ((status != DRV_ERROR_NONE) && (status != DRV_ERROR_QUEUE_RE_SUBSCRIBED)) {
72 1 : BQS_LOG_ERROR("halQueueSubscribe queue[%u] on device[%u] in group[%u] failed, ret=%d.",
73 : queueId, deviceId_, enqueGroupId_, static_cast<int32_t>(status));
74 1 : return BQS_STATUS_DRIVER_ERROR;
75 : }
76 :
77 3 : if ((status == DRV_ERROR_QUEUE_RE_SUBSCRIBED) && (Resubscribe(queueId) != BQS_STATUS_OK)) {
78 1 : BQS_LOG_ERROR("ReSubscribeBuffQueue queue[%u] on device[%u] failed.", queueId, deviceId_);
79 1 : return BQS_STATUS_DRIVER_ERROR;
80 : }
81 :
82 2 : if (!(findRet->second)) {
83 1 : findRet->second = true;
84 1 : StatisticManager::GetInstance().ResumeSubscribe();
85 : }
86 :
87 2 : BQS_LOG_INFO("UpdateSubscribe queue[%u] on device[%u] in group[%u] success.", queueId, deviceId_, enqueGroupId_);
88 2 : return BQS_STATUS_OK;
89 : }
90 :
91 110 : BqsStatus SubscribeManager::Unsubscribe(const uint32_t queueId)
92 : {
93 110 : const auto findRet = subscribeQueuesMaps_.find(queueId);
94 110 : if (findRet == subscribeQueuesMaps_.end()) {
95 1 : BQS_LOG_INFO("queue[%u] devId[%u] is not subscribed, no need to unsubscribe.", queueId, deviceId_);
96 1 : return BQS_STATUS_OK;
97 : }
98 :
99 109 : if (findRet->second) {
100 105 : const auto status = EnhancedUnSubscribe(queueId, QUEUE_ENQUE_EVENT);
101 105 : if (status == DRV_ERROR_NONE) {
102 100 : BQS_LOG_INFO("halQueueUnsubscribe queue[%u] on device[%u] success.", queueId, deviceId_);
103 5 : } else if (status == DRV_ERROR_NOT_EXIST) {
104 1 : BQS_LOG_RUN_WARN("halQueueUnsubscribe return abnormal, queue[%u], device[%u], ret[%d].",
105 : queueId, deviceId_, static_cast<int32_t>(status));
106 : } else {
107 4 : BQS_LOG_ERROR("halQueueUnsubscribe queue[%u] on device[%u] failed, ret=%d.",
108 : queueId, deviceId_, static_cast<int32_t>(status));
109 4 : return BQS_STATUS_DRIVER_ERROR;
110 : }
111 : } else {
112 4 : BQS_LOG_INFO("queue[%u] on device[%u] is pause subscribed, no need to unsubscribe.", queueId, deviceId_);
113 4 : StatisticManager::GetInstance().ResumeSubscribe();
114 : }
115 105 : subscribeQueuesMaps_.erase(findRet);
116 105 : return BQS_STATUS_OK;
117 : }
118 :
119 4 : BqsStatus SubscribeManager::Resubscribe(const uint32_t queueId) const
120 : {
121 4 : auto status = EnhancedUnSubscribe(queueId, QUEUE_ENQUE_EVENT);
122 4 : if ((status != DRV_ERROR_NONE) &&
123 : (status != DRV_ERROR_NOT_EXIST)) {
124 1 : BQS_LOG_ERROR("halQueueUnsubscribe queue[%u] on device[%u] failed, ret=%d.", queueId, deviceId_,
125 : static_cast<int32_t>(status));
126 1 : return BQS_STATUS_DRIVER_ERROR;
127 : }
128 :
129 3 : status = EnhancedSubscribe(queueId, QUEUE_ENQUE_EVENT);
130 3 : if (status != DRV_ERROR_NONE) {
131 2 : BQS_LOG_ERROR("halQueueSubscribe queue[%u] on device[%u] in group[%u] failed, ret=%d.",
132 : queueId, deviceId_, enqueGroupId_, static_cast<int32_t>(status));
133 2 : return BQS_STATUS_DRIVER_ERROR;
134 : }
135 1 : return BQS_STATUS_OK;
136 : }
137 :
138 15 : BqsStatus SubscribeManager::PauseSubscribe(const uint32_t queueId, const uint32_t fullId, const bool idleLog)
139 : {
140 15 : const auto findRet = subscribeQueuesMaps_.find(queueId);
141 15 : if (findRet == subscribeQueuesMaps_.end()) {
142 1 : BQS_LOG_INFO("queue[%u] devId[%u] is not subscribed, no need to pause subscribe.", queueId, deviceId_);
143 1 : return BQS_STATUS_OK;
144 : }
145 :
146 14 : if (findRet->second) {
147 13 : const auto status = EnhancedUnSubscribe(queueId, QUEUE_ENQUE_EVENT);
148 13 : if ((status != DRV_ERROR_NONE) && (status != DRV_ERROR_NOT_EXIST)) {
149 1 : if (idleLog) {
150 1 : BQS_LOG_ERROR("halQueueUnsubscribe queue[%u] on device[%u] failed, ret=%d.",
151 : queueId, deviceId_, static_cast<int32_t>(status));
152 : }
153 1 : return BQS_STATUS_DRIVER_ERROR;
154 : }
155 12 : findRet->second = false;
156 12 : StatisticManager::GetInstance().PauseSubscribe();
157 12 : BQS_LOG_INFO("Pause subscribe queue[%u] on device[%u] as queue or tag[%u] full or get status not success.",
158 : queueId, deviceId_, fullId);
159 : } else {
160 1 : if (idleLog) {
161 1 : BQS_LOG_INFO("queue[%u] on device[%u] is pause subscribed, no need to pause subscribe.",
162 : queueId, deviceId_);
163 : }
164 : }
165 13 : return BQS_STATUS_OK;
166 : }
167 :
168 7 : BqsStatus SubscribeManager::ResumeSubscribe(const uint32_t queueId, const uint32_t notFullId)
169 : {
170 7 : const auto findRet = subscribeQueuesMaps_.find(queueId);
171 7 : if (findRet == subscribeQueuesMaps_.end()) {
172 2 : BQS_LOG_ERROR("queue[%u] devId[%u] is not subscribed, can't resume subscribe.", queueId, deviceId_);
173 2 : return BQS_STATUS_PARAM_INVALID;
174 : }
175 :
176 5 : if (findRet->second) {
177 1 : BQS_LOG_INFO("queue[%u] on device[%u] is subscribed, no need to resume subscribe.", queueId, deviceId_);
178 : } else {
179 4 : const auto status = EnhancedSubscribe(queueId, QUEUE_ENQUE_EVENT);
180 4 : if (status != DRV_ERROR_NONE) {
181 1 : BQS_LOG_ERROR("halQueueSubscribe queue[%u] on device[%u] in group[%u] failed, ret=%d.",
182 : queueId, deviceId_, enqueGroupId_, static_cast<int32_t>(status));
183 1 : return BQS_STATUS_DRIVER_ERROR;
184 : }
185 3 : findRet->second = true;
186 3 : StatisticManager::GetInstance().ResumeSubscribe();
187 3 : BQS_LOG_INFO("Resume subscribe queue[%u] on device[%u] as queue or tag[%u] be not full success.",
188 : queueId, deviceId_, notFullId);
189 : }
190 4 : return BQS_STATUS_OK;
191 : }
192 :
193 156 : BqsStatus SubscribeManager::SubscribeFullToNotFull(uint32_t queueId)
194 : {
195 156 : const auto findRet = fullToNotFullQueuesSets_.find(queueId);
196 156 : if (findRet != fullToNotFullQueuesSets_.end()) {
197 4 : BQS_LOG_INFO("queue[%u] on device[%u] full to not full event is subscribed, no need to unsubscribe.",
198 : queueId, deviceId_);
199 4 : return BQS_STATUS_OK;
200 : }
201 :
202 152 : const auto status = EnhancedSubscribe(queueId, QUEUE_F2NF_EVENT);
203 152 : if ((status != DRV_ERROR_NONE) && (status != DRV_ERROR_QUEUE_RE_SUBSCRIBED)) {
204 1 : BQS_LOG_ERROR("halQueueSubF2NFEvent for queue[%u] in group[%u], deviceId[%u] failed, ret=%d.",
205 : queueId, f2nfGroupId_, deviceId_, static_cast<int32_t>(status));
206 1 : return BQS_STATUS_DRIVER_ERROR;
207 : }
208 :
209 151 : if ((status == DRV_ERROR_QUEUE_RE_SUBSCRIBED) && (ResubscribeF2NF(queueId) != BQS_STATUS_OK)) {
210 2 : BQS_LOG_ERROR("ReHalQueueSubEvent queue[%u] on device[%u] failed.", queueId, deviceId_);
211 2 : return BQS_STATUS_DRIVER_ERROR;
212 : }
213 149 : (void)fullToNotFullQueuesSets_.emplace(queueId);
214 149 : BQS_LOG_INFO("halQueueSubF2NFEvent for queue[%u] in group[%u], deviceId[%u] success.",
215 : queueId, f2nfGroupId_, deviceId_);
216 149 : return BQS_STATUS_OK;
217 : }
218 :
219 5 : BqsStatus SubscribeManager::UpdateSubscribeFullToNotFull(const uint32_t queueId) const
220 : {
221 5 : const auto findRet = fullToNotFullQueuesSets_.find(queueId);
222 5 : if (findRet == fullToNotFullQueuesSets_.end()) {
223 2 : BQS_LOG_INFO("queue[%u] on device[%u] full to not full event is not subscribed, no need to update subscribe.",
224 : queueId, deviceId_);
225 2 : return BQS_STATUS_OK;
226 : }
227 :
228 3 : const auto ret = EnhancedSubscribe(queueId, QUEUE_F2NF_EVENT);
229 3 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_QUEUE_RE_SUBSCRIBED)) {
230 1 : BQS_LOG_ERROR("halQueueSubF2NFEvent for queue[%u] on device[%u] in group[%u] failed, ret=%d.",
231 : queueId, deviceId_, f2nfGroupId_, static_cast<int32_t>(ret));
232 1 : return BQS_STATUS_DRIVER_ERROR;
233 : }
234 :
235 2 : if ((ret == DRV_ERROR_QUEUE_RE_SUBSCRIBED) && (ResubscribeF2NF(queueId) != BQS_STATUS_OK)) {
236 1 : BQS_LOG_ERROR("ReHalQueueSubEvent queue[%u] on device[%u] failed.", queueId, deviceId_);
237 1 : return BQS_STATUS_DRIVER_ERROR;
238 : }
239 1 : BQS_LOG_INFO("UpdateSubscribeFullToNotFull for queue[%u] on device[%u] in group[%u] success.",
240 : queueId, deviceId_, f2nfGroupId_);
241 1 : return BQS_STATUS_OK;
242 : }
243 :
244 134 : BqsStatus SubscribeManager::UnsubscribeFullToNotFull(const uint32_t queueId)
245 : {
246 134 : const auto findRet = fullToNotFullQueuesSets_.find(queueId);
247 134 : if (findRet == fullToNotFullQueuesSets_.end()) {
248 1 : BQS_LOG_INFO("queue[%u] on device[%u] full to not full event is not subscribed, no need to unsubscribe.",
249 : queueId, deviceId_);
250 1 : return BQS_STATUS_OK;
251 : }
252 :
253 133 : const auto status = EnhancedUnSubscribe(queueId, QUEUE_F2NF_EVENT);
254 133 : if (status == DRV_ERROR_NONE) {
255 130 : BQS_LOG_INFO("halQueueUnsubEvent for queue[%u] on device[%u] success.", queueId, deviceId_);
256 3 : } else if ((status == DRV_ERROR_NOT_EXIST) || (status == DRV_ERROR_PERMISSION)) {
257 1 : BQS_LOG_WARN("halQueueUnsubF2NFEvent return abnormal, queue[%u] on device[%u], ret=%d.",
258 : queueId, deviceId_, static_cast<int32_t>(status));
259 : } else {
260 2 : BQS_LOG_ERROR("halQueueUnsubEvent for queue[%u] on device[%u] failed, ret=%d.",
261 : queueId, deviceId_, static_cast<int32_t>(status));
262 2 : return BQS_STATUS_DRIVER_ERROR;
263 : }
264 131 : (void)fullToNotFullQueuesSets_.erase(findRet);
265 131 : return BQS_STATUS_OK;
266 : }
267 :
268 4 : BqsStatus SubscribeManager::ResubscribeF2NF(const uint32_t queueId) const
269 : {
270 4 : auto status = EnhancedUnSubscribe(queueId, QUEUE_F2NF_EVENT);
271 4 : if ((status != DRV_ERROR_NONE) && (status != DRV_ERROR_NOT_EXIST)) {
272 1 : BQS_LOG_ERROR("halQueueUnsubF2NFEvent for queue[%u], deviceId[%u] failed, ret=%d.",
273 : queueId, deviceId_, static_cast<int32_t>(status));
274 1 : return BQS_STATUS_DRIVER_ERROR;
275 : }
276 :
277 3 : status = EnhancedSubscribe(queueId, QUEUE_F2NF_EVENT);
278 3 : if (status != DRV_ERROR_NONE) {
279 2 : BQS_LOG_ERROR("halQueueSubF2NFEvent for queue[%u] in group[%u], deviceId[%u] failed, ret=%d.",
280 : queueId, f2nfGroupId_, deviceId_, static_cast<int32_t>(status));
281 2 : return BQS_STATUS_DRIVER_ERROR;
282 : }
283 1 : return BQS_STATUS_OK;
284 : }
285 :
286 4 : drvError_t SubscribeManager::DefalutSubscribe(const uint32_t queueId, const QUEUE_EVENT_TYPE eventType) const
287 : {
288 4 : if (deviceId_ != dstDeviceId_) {
289 1 : BQS_LOG_ERROR("Subscribe queue[%u] on device[%u] to device[%u] failed because it is not supported.",
290 : queueId, deviceId_, dstDeviceId_);
291 1 : return DRV_ERROR_NOT_SUPPORT;
292 : }
293 :
294 3 : if (eventType == QUEUE_ENQUE_EVENT) {
295 1 : return halQueueSubscribe(deviceId_, queueId, enqueGroupId_, static_cast<int32_t>(QUEUE_TYPE_GROUP));
296 : }
297 :
298 2 : if (eventType == QUEUE_F2NF_EVENT) {
299 1 : return halQueueSubF2NFEvent(deviceId_, queueId, enqueGroupId_);
300 : }
301 :
302 1 : BQS_LOG_ERROR("Invalid event type %d", static_cast<int32_t>(eventType));
303 1 : return DRV_ERROR_INVALID_VALUE;
304 : }
305 :
306 305 : drvError_t SubscribeManager::EnhancedSubscribe(const uint32_t queueId, const QUEUE_EVENT_TYPE eventType) const
307 : {
308 305 : if (extendDriverInterface_) {
309 302 : QueueSubPara queSubParm = {};
310 302 : queSubParm.eventType = eventType;
311 302 : queSubParm.qid = queueId;
312 302 : queSubParm.groupId = enqueGroupId_;
313 302 : queSubParm.devId = deviceId_;
314 302 : queSubParm.dstDevId = dstDeviceId_;
315 302 : queSubParm.flag = QUEUE_SUB_FLAG_SPEC_DST_DEVID;
316 :
317 302 : if (eventType == QUEUE_ENQUE_EVENT) {
318 143 : queSubParm.queType = QUEUE_TYPE_GROUP;
319 : }
320 302 : BQS_LOG_RUN_INFO("Subscribe event[%d] of queue[%u] on device[%u] to group[%u] on resDevice[%u]",
321 : static_cast<int32_t>(eventType), queueId, deviceId_, enqueGroupId_, dstDeviceId_);
322 302 : return halQueueSubEvent(&queSubParm);
323 : }
324 :
325 3 : return DefalutSubscribe(queueId, eventType);
326 : }
327 :
328 3 : drvError_t SubscribeManager::DefalutUnSubscribe(const uint32_t queueId, const QUEUE_EVENT_TYPE eventType) const
329 : {
330 3 : if (eventType == QUEUE_ENQUE_EVENT) {
331 1 : return halQueueUnsubscribe(deviceId_, queueId);
332 : }
333 2 : if (eventType == QUEUE_F2NF_EVENT) {
334 1 : return halQueueUnsubF2NFEvent(deviceId_, queueId);
335 : }
336 :
337 1 : BQS_LOG_ERROR("Invalid event type %d", static_cast<int32_t>(eventType));
338 1 : return DRV_ERROR_INVALID_VALUE;
339 : }
340 :
341 264 : drvError_t SubscribeManager::EnhancedUnSubscribe(const uint32_t queueId, const QUEUE_EVENT_TYPE eventType) const
342 : {
343 264 : if (extendDriverInterface_) {
344 261 : QueueUnsubPara queUnSubParm = {};
345 261 : queUnSubParm.eventType = eventType;
346 261 : queUnSubParm.qid = queueId;
347 261 : queUnSubParm.devId = deviceId_;
348 :
349 261 : return halQueueUnsubEvent(&queUnSubParm);
350 : }
351 :
352 3 : return DefalutUnSubscribe(queueId, eventType);
353 : }
354 :
355 543 : Subscribers &Subscribers::GetInstance()
356 : {
357 543 : static Subscribers instance;
358 543 : return instance;
359 : }
360 :
361 30 : void Subscribers::InitSubscribeManagers(const std::set<uint32_t> &deviceIds, const uint32_t dstDeviceId)
362 : {
363 30 : const uint32_t resId = GlobalCfg::GetInstance().GetResIndexByDeviceId(dstDeviceId);
364 30 : const uint32_t groupId = GlobalCfg::GetInstance().GetGroupIdByDeviceId(dstDeviceId);
365 64 : for (uint32_t resDevId : deviceIds) {
366 34 : subscribeManagers_[resId][resDevId].InitSubscribeManager(resDevId, groupId, groupId, dstDeviceId);
367 : }
368 30 : }
369 :
370 506 : SubscribeManager *Subscribers::GetSubscribeManager(const uint32_t resId, const uint32_t deviceId)
371 : {
372 506 : const auto resIdIter = subscribeManagers_.find(resId);
373 506 : if (resIdIter == subscribeManagers_.end()) {
374 10 : return nullptr;
375 : }
376 496 : const auto deviceIdIter = resIdIter->second.find(deviceId);
377 496 : if (deviceIdIter == resIdIter->second.end()) {
378 0 : return nullptr;
379 : }
380 :
381 496 : return &(deviceIdIter->second);
382 : }
383 :
384 : } // namespace bqs
|
