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 "queue_manager.h"
12 :
13 : #include <cstring>
14 : #include <securec.h>
15 : #include "driver/ascend_hal.h"
16 : #include "bqs_msg.h"
17 : #include "subscribe_manager.h"
18 : #include "statistic_manager.h"
19 : #include "server/bqs_server.h"
20 : #include "router_server.h"
21 : #include "common/bqs_util.h"
22 :
23 : namespace bqs {
24 : namespace {
25 : constexpr const char_t *RELATION_QUEUE_NAME = "QSRelationEvent";
26 : constexpr const char_t *F2NF_QUEUE_NAME = "QSF2NFEvent";
27 : constexpr const char_t *ASYNC_MEM_BUFF_DEQ_QUEUE_NAME = "QSAsyncMemDeBuffEvent";
28 : constexpr const char_t *ASYNC_MEM_BUFF_ENQ_QUEUE_NAME = "QSAsyncMemEnBuffEvent";
29 : // relation message type enqueue message length
30 : constexpr const uint32_t BIND_EVENT_MSG_LENGTH(1U);
31 : constexpr const uint32_t MAX_QUEUE_DEPTH = 8U * 1024U;
32 : constexpr const uint32_t MAX_DEQUEUE_COUNT = MAX_QUEUE_DEPTH;
33 : constexpr const char_t *RELATION_QUEUE_NAME_EXTRA = "QSRelationEventExtra";
34 : constexpr const char_t *F2NF_QUEUE_NAME_EXTRA = "QSF2NFEventExtra";
35 : } // namespace
36 :
37 49 : QueueManager::QueueManager()
38 49 : : deviceId_(0U),
39 49 : groupId_(0U),
40 49 : relationEventQId_(0U),
41 49 : fullToNotFullEventQId_(0U),
42 49 : asyncMemDequeueBuffQId_(0),
43 49 : asyncMemEnqueueBuffQId_(0),
44 49 : initialized_(false),
45 49 : stopped_(false),
46 49 : f2nfQueueEmptyFlag_(true),
47 49 : mbufForF2nf_(nullptr),
48 49 : relationEventQInitialized_(false),
49 49 : fullToNotFullEventQInitialized_(false),
50 49 : deviceIdExtra_(0U),
51 49 : groupIdExtra_(0U),
52 49 : relationEventQIdExtra_(0U),
53 49 : relationEventQInitializedExtra_(false),
54 49 : fullToNotFullEventQIdExtra_(0U),
55 49 : fullToNotFullEventQInitializedExtra_(false),
56 49 : mbufForF2nfExtra_(nullptr),
57 49 : f2nfQueueEmptyFlagExtra_(true),
58 49 : isTriggeredByAsyncMemDequeue_(false),
59 49 : isTriggeredByAsyncMemEnqueue_(false),
60 49 : ayncMemBuffEventQInitialized_(false),
61 98 : initiallizedExtra_(false)
62 49 : {}
63 :
64 49 : QueueManager::~QueueManager()
65 : {
66 49 : Clear();
67 49 : }
68 :
69 256 : QueueManager &QueueManager::GetInstance()
70 : {
71 256 : static QueueManager instance;
72 257 : return instance;
73 : }
74 :
75 206 : BqsStatus QueueManager::CreateQueue(const char_t * const name, const uint32_t depth, uint32_t &queueId,
76 : uint32_t deviceId) const
77 : {
78 206 : std::string nameStr(name);
79 206 : const auto curPid = static_cast<uint32_t>(drvDeviceGetBareTgid());
80 206 : nameStr += std::to_string(curPid);
81 206 : QueueAttr queAttr = {};
82 412 : const auto memcpyRet = memcpy_s(queAttr.name, static_cast<uint32_t>(QUEUE_MAX_STR_LEN),
83 206 : nameStr.c_str(), nameStr.length());
84 206 : if (memcpyRet != EOK) {
85 0 : BQS_LOG_ERROR("CreateAndSubscribeQueue memcpy_s failed, ret[%d].", memcpyRet);
86 0 : return BQS_STATUS_INNER_ERROR;
87 : }
88 206 : queAttr.depth = depth;
89 206 : queAttr.deploy_type = CLIENT_QUEUE_DEPLOY;
90 206 : if (bqs::GetRunContext() == bqs::RunContext::HOST) {
91 184 : queAttr.deploy_type = LOCAL_QUEUE_DEPLOY;
92 : }
93 206 : BQS_LOG_INFO("CreateAndSubscribeQueue queue depth[%d]", depth);
94 206 : const drvError_t ret = halQueueCreate(deviceId, &queAttr, &queueId);
95 206 : if (ret != DRV_ERROR_NONE) {
96 4 : BQS_LOG_ERROR("Create buff queue[%s] error, ret=%d", nameStr.c_str(), static_cast<int32_t>(ret));
97 4 : return BQS_STATUS_DRIVER_ERROR;
98 : }
99 202 : const drvError_t drvRet = halQueueAttach(deviceId, queueId, 0);
100 202 : if (drvRet != DRV_ERROR_NONE) {
101 1 : BQS_LOG_ERROR("Fail to attach queue[%ud], result[%d]", queueId, static_cast<int32_t>(drvRet));
102 1 : (void)DestroyQueue(queueId);
103 1 : return BQS_STATUS_DRIVER_ERROR;
104 : }
105 201 : BQS_LOG_RUN_INFO("Create buff queue[%s] on device[%u] success, queue[%u].", nameStr.c_str(), deviceId, queueId);
106 201 : return BQS_STATUS_OK;
107 206 : }
108 :
109 187 : BqsStatus QueueManager::CreateQueue(const char_t * const name, const uint32_t depth, uint32_t &queueId) const
110 : {
111 187 : return CreateQueue(name, depth, queueId, deviceId_);
112 : }
113 :
114 187 : BqsStatus QueueManager::CreateAndSubscribeQueue(const char_t * const name,
115 : const uint32_t depth, uint32_t &queueId) const
116 : {
117 187 : const BqsStatus ret = CreateQueue(name, depth, queueId);
118 187 : if (ret != BQS_STATUS_OK) {
119 3 : BQS_LOG_ERROR("Create buff queue[name:%s, depth:%u] failed, ret=%d.",
120 : name, depth, static_cast<int32_t>(ret));
121 3 : return ret;
122 : }
123 :
124 184 : drvError_t drvRet = DRV_ERROR_NONE;
125 184 : if (bqs::GetRunContext() != bqs::RunContext::HOST) {
126 20 : drvRet = halQueueSubscribe(deviceId_, queueId, groupId_, QUEUE_TYPE_GROUP);
127 : } else {
128 : struct QueueSubPara queSubParm;
129 164 : queSubParm.eventType = QUEUE_ENQUE_EVENT;
130 164 : queSubParm.qid = queueId;
131 164 : queSubParm.queType = QUEUE_TYPE_GROUP;
132 164 : queSubParm.groupId = groupId_;
133 164 : queSubParm.devId = deviceId_;
134 164 : queSubParm.flag = 0;
135 164 : drvRet = halQueueSubEvent(&queSubParm);
136 : }
137 :
138 184 : if (drvRet != DRV_ERROR_NONE) {
139 2 : BQS_LOG_ERROR("Subscribe buff queue[name:%s, id:%u] failed, deviceId[%u], groupId[%u], ret=%d.",
140 : name, queueId, deviceId_, groupId_, static_cast<int32_t>(drvRet));
141 2 : (void)DestroyQueue(queueId);
142 2 : return BQS_STATUS_DRIVER_ERROR;
143 : }
144 182 : BQS_LOG_INFO("Subscribe buff queue[name:%s, id:%u] success, deviceId[%u], groupId[%u]",
145 : name, queueId, deviceId_, groupId_);
146 182 : return BQS_STATUS_OK;
147 : }
148 :
149 10 : BqsStatus QueueManager::CreateAndSubscribeQueueExtra(const char_t * const name,
150 : const uint32_t depth, uint32_t &queueId) const
151 : {
152 10 : const BqsStatus ret = CreateQueue(name, depth, queueId, deviceIdExtra_);
153 10 : if (ret != BQS_STATUS_OK) {
154 1 : BQS_LOG_ERROR("Create buff queue[name:%s, depth:%u] failed, ret=%d.",
155 : name, depth, static_cast<int32_t>(ret));
156 1 : return ret;
157 : }
158 :
159 9 : const drvError_t drvRet = halQueueSubscribe(deviceIdExtra_, queueId, groupIdExtra_, QUEUE_TYPE_GROUP);
160 9 : if (drvRet != DRV_ERROR_NONE) {
161 1 : BQS_LOG_ERROR("Subscribe buff queue[name:%s, id:%u] failed, deviceId[%u], groupId[%u], ret=%d.",
162 : name, queueId, deviceIdExtra_, groupIdExtra_, static_cast<int32_t>(drvRet));
163 1 : (void)DestroyQueue(queueId, deviceIdExtra_);
164 1 : return BQS_STATUS_DRIVER_ERROR;
165 : }
166 8 : BQS_LOG_INFO("Subscribe buff queue[name:%s, id:%u] success, deviceId[%u], groupId[%u].",
167 : name, queueId, deviceIdExtra_, groupIdExtra_);
168 8 : return BQS_STATUS_OK;
169 : }
170 :
171 189 : BqsStatus QueueManager::DestroyQueue(const uint32_t queueId) const
172 : {
173 189 : return DestroyQueue(queueId, deviceId_);
174 : }
175 :
176 206 : BqsStatus QueueManager::DestroyQueue(const uint32_t queueId, uint32_t deviceId) const
177 : {
178 206 : const int32_t ret = halQueueDestroy(deviceId, queueId);
179 206 : if (ret != static_cast<int32_t>(DRV_ERROR_NONE)) {
180 9 : BQS_LOG_ERROR("halQueueDestroy failed, queue id:[%u], ret:[%d]", queueId, ret);
181 9 : return BQS_STATUS_DRIVER_ERROR;
182 : }
183 197 : BQS_LOG_RUN_INFO("Destroy queue[%u] on deviceId[%u] success.", queueId, deviceId);
184 197 : return BQS_STATUS_OK;
185 : }
186 :
187 187 : BqsStatus QueueManager::UnsubscribeQueue(const uint32_t queueId, const QUEUE_EVENT_TYPE eventType) const
188 : {
189 187 : auto status = DRV_ERROR_NONE;
190 187 : if (bqs::GetRunContext() != bqs::RunContext::HOST) {
191 20 : status = halQueueUnsubscribe(deviceId_, queueId);
192 : } else {
193 : struct QueueUnsubPara queUnsubParm;
194 167 : queUnsubParm.eventType = eventType;
195 167 : queUnsubParm.qid = queueId;
196 167 : queUnsubParm.devId = deviceId_;
197 167 : status = halQueueUnsubEvent(&queUnsubParm);
198 : }
199 :
200 187 : if ((status != DRV_ERROR_NONE) && (status != DRV_ERROR_NOT_EXIST)) {
201 1 : BQS_LOG_ERROR("halQueueUnsubscribe queue[%u] failed, ret=%d.", queueId, static_cast<int32_t>(status));
202 1 : return BQS_STATUS_DRIVER_ERROR;
203 : }
204 186 : return BQS_STATUS_OK;
205 : }
206 :
207 : /**
208 : * init/create/subscribe buff queue
209 : * @return BQS_STATUS_OK:success other:failed
210 : */
211 50 : BqsStatus QueueManager::InitQueueManager(const uint32_t deviceId, const uint32_t groupId, const bool hasAICPU,
212 : const std::string& groupName)
213 : {
214 : (void)hasAICPU;
215 50 : deviceId_ = deviceId;
216 50 : groupId_ = groupId;
217 50 : grpName_ = groupName;
218 50 : BQS_LOG_INFO("QueueManager init begin, deviceId[%u], groupId[%u], groupName[%s].", deviceId, groupId,
219 : grpName_.c_str());
220 :
221 50 : if (!groupName.empty()) {
222 50 : const auto queueInitRet = InitQueue();
223 50 : if (queueInitRet != BQS_STATUS_OK) {
224 6 : BQS_LOG_ERROR("QueueManager Init failed, deviceId[%u], groupId[%u]", deviceId, groupId);
225 6 : return queueInitRet;
226 : }
227 : }
228 :
229 44 : BQS_LOG_INFO("QueueManager init success, deviceId[%u], groupId[%u].", deviceId, groupId);
230 44 : return BQS_STATUS_OK;
231 : }
232 :
233 :
234 5 : void QueueManager::InitExtra(const uint32_t deviceIdExtra, const uint32_t groupIdExtra)
235 : {
236 5 : deviceIdExtra_ = deviceIdExtra;
237 5 : groupIdExtra_ = groupIdExtra;
238 5 : if (!grpName_.empty()) {
239 4 : const auto queueInitRet = InitQueueExtra();
240 4 : if (queueInitRet != BQS_STATUS_OK) {
241 0 : BQS_LOG_ERROR("QueueManager Init failed, deviceId[%u], groupId[%u]", deviceIdExtra, groupIdExtra);
242 : }
243 : }
244 5 : }
245 :
246 7 : BqsStatus QueueManager::InitQueueExtra()
247 : {
248 7 : BQS_LOG_INFO("InitQueueExtra begin, deviceId[%u].", deviceIdExtra_);
249 7 : const auto ret = halQueueInit(deviceIdExtra_);
250 7 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_REPEATED_INIT)) {
251 1 : BQS_LOG_ERROR("halQueueInit error, ret=[%d]", static_cast<int32_t>(ret));
252 1 : return BQS_STATUS_DRIVER_ERROR;
253 : }
254 :
255 6 : BqsStatus bqsRet = CreateAndSubscribeQueueExtra(RELATION_QUEUE_NAME_EXTRA, MAX_QUEUE_DEPTH, relationEventQIdExtra_);
256 6 : if (bqsRet != BQS_STATUS_OK) {
257 1 : BQS_LOG_ERROR("Create and subscribe relation queue error, ret=[%d]", static_cast<int32_t>(bqsRet));
258 1 : return bqsRet;
259 : }
260 5 : relationEventQInitializedExtra_ = true;
261 5 : BQS_LOG_INFO("InitQueue[%u] success, deviceId[%u].", relationEventQIdExtra_, deviceIdExtra_);
262 :
263 5 : bqsRet = CreateAndSubscribeQueueExtra(F2NF_QUEUE_NAME_EXTRA, MAX_QUEUE_DEPTH, fullToNotFullEventQIdExtra_);
264 5 : if (bqsRet != BQS_STATUS_OK) {
265 1 : BQS_LOG_ERROR("Create and subscribe F2NF queue error, ret=[%d]", static_cast<int32_t>(bqsRet));
266 1 : return bqsRet;
267 : }
268 4 : fullToNotFullEventQInitializedExtra_ = true;
269 4 : return BQS_STATUS_OK;
270 : }
271 :
272 54 : BqsStatus QueueManager::InitQueue()
273 : {
274 54 : BQS_LOG_INFO("InitQueue begin, deviceId[%u].", deviceId_);
275 54 : if (bqs::GetRunContext() == bqs::RunContext::HOST) {
276 : // local need queue set
277 : QueueSetInputPara inPutParam;
278 49 : (void)halQueueSet(deviceId_, QUEUE_ENABLE_LOCAL_QUEUE, &inPutParam);
279 : }
280 :
281 54 : const auto ret = halQueueInit(deviceId_);
282 54 : if ((ret != DRV_ERROR_NONE) && (ret != DRV_ERROR_REPEATED_INIT)) {
283 2 : BQS_LOG_ERROR("halQueueInit error, ret=[%d]", static_cast<int32_t>(ret));
284 2 : return BQS_STATUS_DRIVER_ERROR;
285 : }
286 :
287 52 : BqsStatus bqsRet = CreateAndSubscribeQueue(RELATION_QUEUE_NAME, MAX_QUEUE_DEPTH, relationEventQId_);
288 52 : if (bqsRet != BQS_STATUS_OK) {
289 3 : BQS_LOG_ERROR("Create and subscribe relation queue error, ret=[%d]", static_cast<int32_t>(bqsRet));
290 3 : return bqsRet;
291 : }
292 49 : relationEventQInitialized_ = true;
293 :
294 49 : bqsRet = CreateAndSubscribeQueue(F2NF_QUEUE_NAME, MAX_QUEUE_DEPTH, fullToNotFullEventQId_);
295 49 : if (bqsRet != BQS_STATUS_OK) {
296 2 : BQS_LOG_ERROR("Create and subscribe F2NF queue error, ret=[%d]", static_cast<int32_t>(bqsRet));
297 2 : return bqsRet;
298 : }
299 47 : fullToNotFullEventQInitialized_ = true;
300 :
301 47 : bqsRet = CreateAndSubscribeQueue(ASYNC_MEM_BUFF_DEQ_QUEUE_NAME, MAX_QUEUE_DEPTH, asyncMemDequeueBuffQId_);
302 47 : if (bqsRet != BQS_STATUS_OK) {
303 1 : BQS_LOG_ERROR("Create and subscribe AsyncMemBuff queue error, ret=[%d]", static_cast<int32_t>(bqsRet));
304 1 : return bqsRet;
305 : }
306 :
307 46 : bqsRet = CreateAndSubscribeQueue(ASYNC_MEM_BUFF_ENQ_QUEUE_NAME, MAX_QUEUE_DEPTH, asyncMemEnqueueBuffQId_);
308 46 : if (bqsRet != BQS_STATUS_OK) {
309 1 : BQS_LOG_ERROR("Create and subscribe AsyncMemBuff queue error, ret=[%d]", static_cast<int32_t>(bqsRet));
310 1 : return bqsRet;
311 : }
312 45 : ayncMemBuffEventQInitialized_ = true;
313 :
314 45 : BQS_LOG_INFO("InitQueue success, deviceId[%u]", deviceId_);
315 45 : return BQS_STATUS_OK;
316 : }
317 :
318 : /**
319 : * destroy buff queue
320 : * @return NA
321 : */
322 60 : void QueueManager::Destroy()
323 : {
324 60 : BQS_LOG_INFO("QueueManager Destroy begin");
325 :
326 : {
327 60 : const std::unique_lock<std::mutex> destroyLock(mutex_);
328 60 : stopped_ = true;
329 60 : cv_.notify_all();
330 60 : }
331 :
332 60 : Clear();
333 60 : BQS_LOG_INFO("QueueManager Destroy success");
334 60 : return;
335 : }
336 :
337 111 : void QueueManager::Clear()
338 : {
339 111 : if (mbufForF2nf_ != nullptr) {
340 1 : (void)halMbufFree(mbufForF2nf_);
341 1 : mbufForF2nf_ = nullptr;
342 : }
343 :
344 111 : if (mbufForF2nfExtra_ != nullptr) {
345 1 : (void)halMbufFree(mbufForF2nfExtra_);
346 1 : mbufForF2nfExtra_ = nullptr;
347 : }
348 :
349 111 : if (relationEventQInitialized_) {
350 49 : ClearQueue(relationEventQId_, QUEUE_ENQUE_EVENT);
351 49 : relationEventQInitialized_ = false;
352 : }
353 :
354 111 : if (relationEventQInitializedExtra_) {
355 5 : halQueueUnsubscribe(deviceIdExtra_, relationEventQIdExtra_);
356 5 : (void)DestroyQueue(relationEventQIdExtra_, deviceIdExtra_);
357 5 : relationEventQInitializedExtra_ = false;
358 : }
359 :
360 111 : if (fullToNotFullEventQInitialized_) {
361 47 : ClearQueue(fullToNotFullEventQId_, QUEUE_F2NF_EVENT);
362 47 : fullToNotFullEventQInitialized_ = false;
363 : }
364 :
365 111 : if (fullToNotFullEventQInitializedExtra_) {
366 4 : halQueueUnsubscribe(deviceIdExtra_, fullToNotFullEventQIdExtra_);
367 4 : (void)DestroyQueue(fullToNotFullEventQIdExtra_, deviceIdExtra_);
368 4 : fullToNotFullEventQInitializedExtra_ = false;
369 : }
370 :
371 111 : if (ayncMemBuffEventQInitialized_) {
372 45 : ClearQueue(asyncMemDequeueBuffQId_, QUEUE_ENQUE_EVENT);
373 45 : ClearQueue(asyncMemEnqueueBuffQId_, QUEUE_ENQUE_EVENT);
374 45 : ayncMemBuffEventQInitialized_ = false;
375 : }
376 111 : }
377 :
378 186 : void QueueManager::ClearQueue(const uint32_t queueId, const QUEUE_EVENT_TYPE eventType) const
379 : {
380 186 : (void)UnsubscribeQueue(queueId, eventType);
381 186 : (void)DestroyQueue(queueId);
382 186 : }
383 :
384 : /**
385 : * work thread init success will notify queue manager
386 : * @return NA
387 : */
388 164 : void QueueManager::NotifyInitSuccess(const uint32_t index)
389 : {
390 164 : const std::unique_lock<std::mutex> notifyLock(mutex_);
391 164 : if (!initialized_ && (index == 0U)) {
392 50 : initialized_ = true;
393 50 : cv_.notify_all();
394 50 : BQS_LOG_INFO("Queue schedule init success.");
395 114 : } else if (!initiallizedExtra_ && (index == 1U)) {
396 1 : initiallizedExtra_ = true;
397 1 : cv_.notify_all();
398 1 : BQS_LOG_INFO("Queue schedule extra init success.");
399 : }
400 328 : return;
401 164 : }
402 :
403 5 : BqsStatus QueueManager::EnqueueRelationEvent()
404 : {
405 : {
406 : // wait for work thread halEschedWaitEvent execute
407 5 : std::unique_lock<std::mutex> equeueRelationEventLock(mutex_);
408 6 : while ((!initialized_) && (!stopped_)) {
409 1 : BQS_LOG_INFO("Relation msg enqueue wait for init success.");
410 1 : cv_.wait(equeueRelationEventLock);
411 1 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
412 : }
413 :
414 5 : if (stopped_) {
415 1 : BQS_LOG_INFO("Queue manager has been stopped, no need to enqueue relation.");
416 1 : return BQS_STATUS_OK;
417 : }
418 5 : }
419 :
420 4 : return EnqueueRelationEventToQ(deviceId_, relationEventQId_);
421 : }
422 :
423 1 : BqsStatus QueueManager::EnqueueRelationEventExtra()
424 : {
425 : {
426 : // wait for work thread halEschedWaitEvent execute
427 1 : std::unique_lock<std::mutex> equeueRelationEventLock(mutex_);
428 1 : while ((!initiallizedExtra_) && (!stopped_)) {
429 0 : BQS_LOG_INFO("Relation msg enqueue wait for init success.");
430 0 : cv_.wait(equeueRelationEventLock);
431 0 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
432 : }
433 :
434 1 : if (stopped_) {
435 1 : BQS_LOG_INFO("Queue manager has been stopped, no need to enqueue relation.");
436 1 : return BQS_STATUS_OK;
437 : }
438 1 : }
439 0 : return EnqueueRelationEventToQ(deviceIdExtra_, relationEventQIdExtra_);
440 : }
441 : /**
442 : * Enqueue a data to implies that the client sent a message
443 : * @return BQS_STATUS_OK:success other:failed
444 : */
445 4 : BqsStatus QueueManager::EnqueueRelationEventToQ(const uint32_t deviceId, const uint32_t relationEventQ) const
446 : {
447 4 : BQS_LOG_INFO("QueueManager EnqueueRelationEvent begin");
448 :
449 4 : Mbuf *mbufPtr = nullptr;
450 4 : int32_t ret = halMbufAlloc(BIND_EVENT_MSG_LENGTH, &mbufPtr);
451 4 : if (ret != DRV_ERROR_NONE) {
452 1 : BQS_LOG_ERROR("halMbufAlloc error, queue id:[%u], ret=[%d]", relationEventQ, ret);
453 1 : return BQS_STATUS_DRIVER_ERROR;
454 : }
455 :
456 3 : ret = halQueueEnQueue(deviceId, relationEventQ, mbufPtr);
457 3 : if (ret != DRV_ERROR_NONE) {
458 1 : BQS_LOG_ERROR("halQueueEnQueue error, queue id:[%u], ret=[%d]", relationEventQ, ret);
459 1 : (void)halMbufFree(mbufPtr);
460 1 : return BQS_STATUS_DRIVER_ERROR;
461 : }
462 2 : StatisticManager::GetInstance().RelationEnqueueStat();
463 2 : BQS_LOG_INFO("QueueManager EnqueueRelationEvent end, queueId[%u] deviceId[%u]", relationEventQ, deviceId);
464 2 : return BQS_STATUS_OK;
465 : }
466 :
467 : /**
468 : * handle the bind or unbind msg that the client sent
469 : * @return true:has handle relation msg, false:not handle
470 : */
471 6 : bool QueueManager::HandleRelationEvent(const uint32_t index) const
472 : {
473 6 : bool dequeue = false;
474 6 : Mbuf *mbufPtr = nullptr;
475 6 : int32_t ret = DRV_ERROR_NONE;
476 6 : auto deviceId = (index == 0U) ? deviceId_ : deviceIdExtra_;
477 6 : auto relationEventQId = (index == 0U) ? relationEventQId_ : relationEventQIdExtra_;
478 : while (true) {
479 9 : ret = halQueueDeQueue(deviceId, relationEventQId, PtrToPtr< Mbuf *, void *>(&mbufPtr));
480 9 : if (ret == DRV_ERROR_QUEUE_EMPTY) {
481 5 : break;
482 : }
483 :
484 4 : BQS_LOG_INFO("HandleRelationEvent dequeue deviceId[%u], relationQ[%u]", deviceId, relationEventQId);
485 4 : if (ret != DRV_ERROR_NONE) {
486 1 : BQS_LOG_ERROR("halQueueDeQueue error, queue id:[%u], ret=[%d]", relationEventQId, ret);
487 1 : break;
488 : }
489 3 : dequeue = true;
490 3 : StatisticManager::GetInstance().RelationDequeueStat();
491 :
492 3 : ret = halMbufFree(mbufPtr);
493 3 : if (ret != DRV_ERROR_NONE) {
494 1 : BQS_LOG_ERROR("halMbufFree error, queue id:[%u], ret=[%d]", relationEventQId, ret);
495 : }
496 : }
497 :
498 6 : if (dequeue) {
499 : // pipeline queue id is valid : event mode
500 3 : if (RouterServer::GetInstance().GetPipelineQueueId() < MAX_QUEUE_ID_NUM) {
501 0 : RouterServer::GetInstance().BindMsgProc(index);
502 : } else {
503 3 : BqsServer::GetInstance().BindMsgProc();
504 : }
505 3 : BQS_LOG_INFO("HandleRelationEvent end.");
506 : }
507 6 : return dequeue;
508 : }
509 :
510 4 : void QueueManager::MakeUpMbuf(Mbuf **mbufPtr) const
511 : {
512 : // malloc mbuf for f2nf queue
513 4 : const size_t mbufLen = sizeof(EntityInfo);
514 4 : auto drvRet = halMbufAlloc(mbufLen, mbufPtr);
515 4 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
516 1 : BQS_LOG_ERROR("Failed to malloc mbuf, ret=[%d]", drvRet);
517 1 : return;
518 : }
519 3 : Mbuf* &mbuf = *mbufPtr;
520 3 : drvRet = halMbufSetDataLen(mbuf, mbufLen);
521 3 : if (drvRet != static_cast<int32_t>(DRV_ERROR_NONE)) {
522 1 : BQS_LOG_ERROR("Set data len for mbuf failed, dataLen:[%zu], ret=[%d]", mbufLen, drvRet);
523 1 : (void)halMbufFree(mbuf);
524 1 : mbuf = nullptr;
525 1 : return;
526 : }
527 : }
528 :
529 5 : void QueueManager::MakeUpF2NFMbuf(const uint32_t index)
530 : {
531 5 : if (index == 0U) {
532 : static std::once_flag onceFlag;
533 4 : std::call_once(onceFlag, [&]() {
534 1 : BQS_LOG_INFO("make up mbufForF2nf_");
535 1 : MakeUpMbuf(&mbufForF2nf_);
536 1 : });
537 : } else {
538 : static std::once_flag onceFlagExtra;
539 1 : std::call_once(onceFlagExtra, [&]() {
540 1 : BQS_LOG_INFO("make up mbufForF2nfExtra_");
541 1 : MakeUpMbuf(&mbufForF2nfExtra_);
542 1 : });
543 : }
544 5 : }
545 : /**
546 : * enqueue the queue id of full to not full queue
547 : * @return BQS_STATUS_OK:success other:failed
548 : */
549 5 : BqsStatus QueueManager::EnqueueFullToNotFullEvent(const uint32_t index)
550 : {
551 5 : MakeUpF2NFMbuf(index);
552 5 : auto &f2nfLock = (index == 0U) ? f2nfLock_ : f2nfLockExtra_;
553 5 : auto &mbufForF2nf = (index == 0U) ? mbufForF2nf_ : mbufForF2nfExtra_;
554 5 : auto &f2nfQueueEmptyFlag = (index == 0U) ? f2nfQueueEmptyFlag_ : f2nfQueueEmptyFlagExtra_;
555 5 : auto &deviceId = (index == 0U) ? deviceId_ : deviceIdExtra_;
556 5 : auto &fullToNotFullEventQId = (index == 0U) ? fullToNotFullEventQId_ : fullToNotFullEventQIdExtra_;
557 :
558 5 : if (mbufForF2nf == nullptr) {
559 2 : BQS_LOG_ERROR("Failed to malloc mbuf for f2nf event, index is %u.", index);
560 2 : return BqsStatus::BQS_STATUS_INNER_ERROR;
561 : }
562 :
563 3 : f2nfLock.Lock();
564 3 : if (!f2nfQueueEmptyFlag.load()) {
565 1 : DGW_LOG_DEBUG("EnqueueFullToNotFullEvent f2NfQueueEmptyFlag is false, index is %u.", index);
566 1 : f2nfLock.Unlock();
567 1 : return BqsStatus::BQS_STATUS_OK;
568 : }
569 2 : BQS_LOG_INFO("Begin to enqueue f2nf event, index is %u.", index);
570 : // In order to prevent event from being dropped, first store f2nFQueueEmptyFlag to false, then enqueue
571 2 : f2nfQueueEmptyFlag.store(false);
572 2 : const auto ret = halQueueEnQueue(deviceId, fullToNotFullEventQId, mbufForF2nf);
573 2 : if (ret != DRV_ERROR_NONE) {
574 : // reset f2nfQueueEmptyFlag, perhaps events were dropped because enqueue failed
575 1 : f2nfQueueEmptyFlag.store(true);
576 1 : BQS_LOG_ERROR("halQueueEnQueue error, ret=[%d], index is %u.", ret, index);
577 1 : f2nfLock.Unlock();
578 1 : return BqsStatus::BQS_STATUS_DRIVER_ERROR;
579 : }
580 1 : f2nfLock.Unlock();
581 1 : StatisticManager::GetInstance().F2nfEnqueueStat();
582 1 : BQS_LOG_INFO("Finish to enqueue f2nf event, index is %u,", index);
583 1 : return BqsStatus::BQS_STATUS_OK;
584 : }
585 :
586 : /**
587 : * handle the event of full to not full
588 : * @return true:has handle f2nf msg, false:not handle
589 : */
590 6 : bool QueueManager::HandleFullToNotFullEvent(const uint32_t index)
591 : {
592 6 : uint32_t dequeueCount = 0U;
593 6 : auto &f2nfLock = (index == 0U) ? f2nfLock_ : f2nfLockExtra_;
594 6 : auto &mbufForF2nf = (index == 0U) ? mbufForF2nf_ : mbufForF2nfExtra_;
595 6 : auto &f2nfQueueEmptyFlag = (index == 0U) ? f2nfQueueEmptyFlag_ : f2nfQueueEmptyFlagExtra_;
596 6 : auto &deviceId = (index == 0U) ? deviceId_ : deviceIdExtra_;
597 6 : auto &fullToNotFullEventQId = (index == 0U) ? fullToNotFullEventQId_ : fullToNotFullEventQIdExtra_;
598 :
599 6 : f2nfLock.Lock();
600 10 : while (dequeueCount <= MAX_DEQUEUE_COUNT) {
601 10 : void *dequeuedMbuf = nullptr;
602 10 : const int32_t ret = halQueueDeQueue(deviceId, fullToNotFullEventQId, &dequeuedMbuf);
603 10 : if (ret == DRV_ERROR_QUEUE_EMPTY) {
604 5 : break;
605 : }
606 5 : if (ret != DRV_ERROR_NONE) {
607 1 : BQS_LOG_ERROR("halQueueDeQueue error, queue id:[%u], ret=[%d]", fullToNotFullEventQId, ret);
608 1 : f2nfLock.Unlock();
609 1 : return false;
610 : }
611 4 : mbufForF2nf = PtrToPtr<void, Mbuf>(dequeuedMbuf);
612 4 : StatisticManager::GetInstance().F2nfDequeueStat();
613 : // mbuf for f2nf event will be freed when destroy queue manager
614 4 : dequeueCount++;
615 : }
616 5 : f2nfQueueEmptyFlag.store(true);
617 5 : f2nfLock.Unlock();
618 5 : return (dequeueCount > 0U);
619 : }
620 :
621 : /**
622 : * Enqueue a data to implies that the clientQ sent a message
623 : * @return BQS_STATUS_OK:success other:failed
624 : */
625 5 : BqsStatus QueueManager::EnqueueAsynMemBuffEvent()
626 : {
627 : // avoid parallel halEnQueue
628 5 : std::lock_guard<std::mutex> lock(mutex_);
629 5 : Mbuf *mbufPtr = nullptr;
630 5 : int32_t ret = halMbufAlloc(BIND_EVENT_MSG_LENGTH, &mbufPtr);
631 5 : if (ret != DRV_ERROR_NONE) {
632 1 : BQS_LOG_ERROR("halMbufAlloc error, queue id:[%u]/[%u], ret=[%d]",
633 : asyncMemDequeueBuffQId_, asyncMemEnqueueBuffQId_, ret);
634 1 : return BQS_STATUS_DRIVER_ERROR;
635 : }
636 :
637 4 : uint32_t asyncMemBuffQId = 0;
638 4 : if (isTriggeredByAsyncMemDequeue_) {
639 2 : asyncMemBuffQId = asyncMemDequeueBuffQId_;
640 2 : ret = halQueueEnQueue(deviceId_, asyncMemDequeueBuffQId_, mbufPtr);
641 2 : } else if (isTriggeredByAsyncMemEnqueue_) {
642 2 : asyncMemBuffQId = asyncMemEnqueueBuffQId_;
643 2 : ret = halQueueEnQueue(deviceId_, asyncMemEnqueueBuffQId_, mbufPtr);
644 : } else {
645 : ;
646 : }
647 :
648 4 : if (ret != DRV_ERROR_NONE) {
649 2 : BQS_LOG_ERROR("halQueueEnQueue error, queue id:[%u], ret=[%d]", asyncMemBuffQId, ret);
650 2 : (void)halMbufFree(mbufPtr);
651 2 : return BQS_STATUS_DRIVER_ERROR;
652 : }
653 2 : StatisticManager::GetInstance().AsynMemEnqueueStat();
654 2 : return BQS_STATUS_OK;
655 5 : }
656 :
657 : /**
658 : * handle the event of Aysn mem buffer
659 : * @return true:has handle Aysn mem buffer msg, false:not handle
660 : */
661 5 : bool QueueManager::HandleAsynMemBuffEvent(const uint32_t index)
662 : {
663 : (void)index;
664 5 : Mbuf *mbufPtr = nullptr;
665 5 : int32_t ret = DRV_ERROR_RESERVED;
666 5 : bool isTriggered = false;
667 : while (true) {
668 6 : uint32_t asyncMemBuffQId = 0;
669 6 : if (isTriggeredByAsyncMemDequeue_) {
670 5 : isTriggered = true;
671 5 : asyncMemBuffQId = asyncMemDequeueBuffQId_;
672 5 : ret = halQueueDeQueue(deviceId_, asyncMemDequeueBuffQId_, PtrToPtr<Mbuf *, void *>(&mbufPtr));
673 : }
674 :
675 6 : if (isTriggeredByAsyncMemEnqueue_) {
676 5 : isTriggered = true;
677 5 : asyncMemBuffQId = asyncMemEnqueueBuffQId_;
678 5 : ret = halQueueDeQueue(deviceId_, asyncMemEnqueueBuffQId_, PtrToPtr<Mbuf *, void *>(&mbufPtr));
679 : }
680 :
681 6 : if (ret == DRV_ERROR_QUEUE_EMPTY || ret == DRV_ERROR_RESERVED) {
682 2 : if (isTriggeredByAsyncMemDequeue_) {
683 1 : isTriggeredByAsyncMemDequeue_ = false;
684 : }
685 2 : if (isTriggeredByAsyncMemEnqueue_) {
686 1 : isTriggeredByAsyncMemEnqueue_ = false;
687 : }
688 2 : break;
689 : }
690 :
691 4 : if (ret != DRV_ERROR_NONE) {
692 3 : BQS_LOG_ERROR("halQueueDeQueue error, queue id:[%u], ret=[%d]", asyncMemBuffQId, ret);
693 3 : break;
694 : }
695 :
696 1 : if (isTriggered) {
697 1 : StatisticManager::GetInstance().AsynMemDequeueStat();
698 1 : ret = halMbufFree(mbufPtr);
699 1 : if (ret != DRV_ERROR_NONE) {
700 1 : BQS_LOG_ERROR("halMbufFree error, queue id:[%u], ret=[%d]", asyncMemBuffQId, ret);
701 : }
702 : }
703 1 : }
704 5 : return true;
705 : }
706 :
707 1 : void QueueManager::LogErrorRelationQueueStatus() const
708 : {
709 1 : LogErrorQueueStatus(relationEventQId_);
710 1 : }
711 :
712 1 : void QueueManager::LogErrorQueueStatus(const uint32_t queueId) const
713 : {
714 : QueueInfo queueInfoObj;
715 1 : auto ret = halQueueQueryInfo(deviceId_, queueId, &queueInfoObj);
716 1 : if (ret == DRV_ERROR_NONE) {
717 1 : BQS_LOG_ERROR("halQueueQueryInfo get queue info, deviceId_:%u, "
718 : "queueId:%u, size:%d, depth:%d, status:%d, workMode:%d, "
719 : "type:%d, subGroupId:%d,subPid:%d, subF2NFGroupId:%d, "
720 : "subF2NFPid:%d, enqueCnt:%llu, dequeCnt:%llu, "
721 : "enqueFailCnt:%llu, dequeFailCnt:%llu, enqueEventOk:%llu, "
722 : "enqueEventFail:%llu, f2nfEventOk:%llu, f2nfEventFail:%llu, "
723 : "lastEnqueTime.tv_sec:%ld, lastEnqueTime.tv_usec:%ld, "
724 : "lastDequeTime.tv_sec:%ld, lastDequeTime.tv_usec:%ld.",
725 : deviceId_, queueInfoObj.id, queueInfoObj.size, queueInfoObj.depth,
726 : queueInfoObj.status, queueInfoObj.workMode, queueInfoObj.type,
727 : queueInfoObj.subGroupId, queueInfoObj.subPid, queueInfoObj.subF2NFGroupId,
728 : queueInfoObj.subF2NFPid, queueInfoObj.stat.enqueCnt,
729 : queueInfoObj.stat.dequeCnt, queueInfoObj.stat.enqueFailCnt,
730 : queueInfoObj.stat.dequeFailCnt, queueInfoObj.stat.enqueEventOk,
731 : queueInfoObj.stat.enqueEventFail, queueInfoObj.stat.f2nfEventOk,
732 : queueInfoObj.stat.f2nfEventFail, queueInfoObj.stat.lastEnqueTime.tv_sec,
733 : queueInfoObj.stat.lastEnqueTime.tv_usec, queueInfoObj.stat.lastDequeTime.tv_sec,
734 : queueInfoObj.stat.lastDequeTime.tv_usec);
735 : } else {
736 0 : BQS_LOG_ERROR("halQueueQueryInfo error, deviceId_:[%u], queueId:[%u], ret:[%d]", deviceId_, queueId, ret);
737 : }
738 :
739 1 : int32_t status = QUEUE_NORMAL;
740 1 : ret = halQueueGetStatus(deviceId_, queueId, QUERY_QUEUE_STATUS, static_cast<uint32_t>(sizeof(uint32_t)), &status);
741 1 : if (ret == DRV_ERROR_NONE) {
742 1 : BQS_LOG_DEBUG("halQueueGetStatus succ, queueId:[%u] status:[%d].", queueId, status);
743 : } else {
744 0 : BQS_LOG_ERROR("halQueueGetStatus failed, queueId:[%u], ret:[%d].", queueId, ret);
745 : }
746 1 : }
747 : } // namespace bqs
|
