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