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