444 lines
10 KiB
C++
444 lines
10 KiB
C++
/*
|
|
* Copyright (C) 2021 Amlogic Corporation.
|
|
*
|
|
* Licensed under the Apache License, Version 2.0 (the "License");
|
|
* you may not use this file except in compliance with the License.
|
|
* You may obtain a copy of the License at
|
|
*
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
*
|
|
* Unless required by applicable law or agreed to in writing, software
|
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
* See the License for the specific language governing permissions and
|
|
* limitations under the License.
|
|
*/
|
|
#include <unistd.h>
|
|
#include "Queue.h"
|
|
|
|
namespace ATools {
|
|
|
|
Queue::Queue()
|
|
{
|
|
_init();
|
|
}
|
|
|
|
Queue::Queue(uint32_t maxElement)
|
|
{
|
|
_init();
|
|
if (maxElement > 0) {
|
|
mCapability = maxElement;
|
|
}
|
|
}
|
|
|
|
Queue::Queue(bool ascendingOrder, int (*cmp)(void *, void *))
|
|
{
|
|
_init();
|
|
if (cmp) {
|
|
mSort = true;
|
|
mAscendingOrder = ascendingOrder;
|
|
cmpEleFun = cmp;
|
|
}
|
|
}
|
|
|
|
Queue::Queue(uint32_t maxElement, bool ascendingOrder, int (*cmp)(void *, void *))
|
|
{
|
|
_init();
|
|
if (maxElement > 0) {
|
|
mCapability = maxElement;
|
|
}
|
|
if (cmp) {
|
|
mSort = true;
|
|
mAscendingOrder = ascendingOrder;
|
|
cmpEleFun = cmp;
|
|
}
|
|
}
|
|
|
|
Queue::~Queue()
|
|
{
|
|
_release();
|
|
}
|
|
|
|
int32_t Queue::push(void *ele)
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _pushElement(ele, false);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return ret;
|
|
}
|
|
|
|
int32_t Queue::pushAndWait(void *ele)
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _pushElement(ele, true);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return ret;
|
|
}
|
|
|
|
int32_t Queue::pop(void **e)
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _popElement(e, false, NULL, NULL);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return ret;
|
|
}
|
|
|
|
int32_t Queue::popAndWait(void **e)
|
|
{
|
|
*e = NULL;
|
|
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _popElement(e, true, NULL, NULL);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return ret;
|
|
}
|
|
|
|
int32_t Queue::peek(void **e, int32_t pos)
|
|
{
|
|
*e = NULL;
|
|
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _peekElement(e, pos);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return ret;
|
|
}
|
|
|
|
|
|
int32_t Queue::flush()
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _flushElements(NULL, NULL);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
return ret;
|
|
}
|
|
int32_t Queue::flushAndCallback(void *userdata, void (*fcb)(void *userdata, void *ele))
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
int32_t ret = _flushElements(userdata, fcb);
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return ret;
|
|
}
|
|
|
|
|
|
int32_t Queue::getCnt()
|
|
{
|
|
int32_t cnt = 0;
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
cnt = mUsedElementCnts;
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return cnt;
|
|
}
|
|
|
|
|
|
bool Queue::isEmpty()
|
|
{
|
|
bool isEmpty;
|
|
|
|
_lock();
|
|
|
|
if (mUsedFirstElement == NULL || mUsedLastElement == NULL)
|
|
isEmpty = true;
|
|
else
|
|
isEmpty = false;
|
|
|
|
_unlock();
|
|
|
|
return isEmpty;
|
|
}
|
|
|
|
|
|
int32_t Queue::setAllowedNewData(bool allowed)
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
mAllowedNewData = allowed;
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
if (mAllowedNewData == false) {
|
|
// notify waiting threads, when new data isn't accepted
|
|
pthread_cond_broadcast(&mCondGet);
|
|
pthread_cond_broadcast(&mCondPut);
|
|
}
|
|
|
|
return Q_OK;
|
|
}
|
|
|
|
bool Queue::isAllowedNewData()
|
|
{
|
|
if (Q_OK != _lock())
|
|
return Q_ERR_LOCK;
|
|
|
|
bool allowed = mAllowedNewData;
|
|
|
|
if (Q_OK != _unlock())
|
|
return Q_ERR_LOCK;
|
|
|
|
return allowed;
|
|
}
|
|
|
|
|
|
int32_t Queue::_init()
|
|
{
|
|
pthread_mutex_init(&mMutex, NULL);
|
|
|
|
pthread_cond_init(&mCondGet, NULL);
|
|
|
|
pthread_cond_init(&mCondPut, NULL);
|
|
|
|
mUsedFirstElement = NULL;
|
|
mUsedLastElement = NULL;
|
|
mUsedElementCnts = 0;
|
|
mCapability = INT32_MAX - 1;//max capability
|
|
mAllowedNewData = true;
|
|
mSort = false;
|
|
mAscendingOrder = 1;
|
|
cmpEleFun = NULL;
|
|
return Q_OK;
|
|
}
|
|
|
|
int32_t Queue::_release()
|
|
{
|
|
// this method will not immediately return on error,
|
|
// it will try to release all the memory that was allocated.
|
|
int error = Q_OK;
|
|
// make sure no new data comes and wake all waiting threads
|
|
error = setAllowedNewData(false);
|
|
error = _lock();
|
|
|
|
error = _flushElements(NULL, NULL);
|
|
|
|
mUsedFirstElement = NULL;
|
|
mUsedLastElement = NULL;
|
|
|
|
// destroy lock and queue etc
|
|
error = pthread_cond_destroy(&mCondGet);
|
|
error = pthread_cond_destroy(&mCondPut);
|
|
|
|
error = _unlock();
|
|
while (EBUSY == (error = pthread_mutex_destroy(&mMutex)))
|
|
usleep(100*1000);
|
|
|
|
return error;
|
|
}
|
|
|
|
int32_t Queue::_lock()
|
|
{
|
|
// all errors are unrecoverable for us
|
|
if (0 != pthread_mutex_lock(&mMutex))
|
|
return Q_ERR_LOCK;
|
|
return Q_OK;
|
|
}
|
|
int32_t Queue::_unlock()
|
|
{
|
|
// all errors are unrecoverable for us
|
|
if (0 != pthread_mutex_unlock(&mMutex))
|
|
return Q_ERR_LOCK;
|
|
return Q_OK;
|
|
}
|
|
|
|
int32_t Queue::_flushElements(void *userdata, void (*fcb)(void *userdata, void *ele))
|
|
{
|
|
Element *ele = NULL;
|
|
while (mUsedFirstElement != NULL) {
|
|
ele = mUsedFirstElement;
|
|
mUsedFirstElement = mUsedFirstElement->next;
|
|
if (fcb != NULL) {
|
|
fcb(userdata,ele->data);
|
|
}
|
|
free(ele);
|
|
}
|
|
mUsedFirstElement = NULL;
|
|
mUsedLastElement = NULL;
|
|
mUsedElementCnts = 0;
|
|
|
|
return Q_OK;
|
|
}
|
|
|
|
int32_t Queue::_pushElement(void *ele, bool isWait)
|
|
{
|
|
Element *newEle = NULL;
|
|
if (mAllowedNewData == false) { // no new data allowed
|
|
return Q_ERR_NONEWDATA;
|
|
}
|
|
|
|
// max_elements already reached?
|
|
// if condition _needs_ to be in sync with while loop below!
|
|
if (mUsedElementCnts == mCapability) {
|
|
if (isWait == false) {
|
|
return Q_ERR_NUM_ELEMENTS;
|
|
} else {
|
|
while ((mUsedElementCnts == mCapability) && mAllowedNewData) {
|
|
pthread_cond_wait(&mCondPut, &mMutex);
|
|
}
|
|
if (mAllowedNewData == false) {
|
|
return Q_ERR_NONEWDATA;
|
|
}
|
|
}
|
|
}
|
|
|
|
newEle = (Element *)calloc(1,sizeof(Element));
|
|
if (newEle == NULL) { // could not allocate memory for new elements
|
|
return Q_ERR_MEM;
|
|
}
|
|
newEle->data = ele;
|
|
newEle->next = NULL;
|
|
//printf("_pushElement,%p,%p,data:%p\n",mUsedLastElement,newEle,ele);
|
|
if (mSort == false || mUsedFirstElement == NULL) {
|
|
// insert at the end when we don't want to sort or the queue is empty
|
|
if (mUsedLastElement == NULL)
|
|
mUsedFirstElement = newEle;
|
|
else
|
|
mUsedLastElement->next = newEle;
|
|
mUsedLastElement = newEle;
|
|
} else {
|
|
// search appropriate place to sort element in
|
|
Element *s = mUsedFirstElement; // s != NULL, because of if condition above
|
|
Element *t = NULL;
|
|
//check if insert new element to the first element
|
|
int asc_first_el = (mAscendingOrder == true && cmpEleFun(s->data, ele) >= 0);
|
|
int desc_first_el = (mAscendingOrder == false && cmpEleFun(s->data, ele) <= 0);
|
|
|
|
if (asc_first_el == 0 && desc_first_el == 0) {
|
|
// element will be inserted between s and t
|
|
for (s = mUsedFirstElement, t = s->next; s != NULL && t != NULL; s = t, t = t->next) {
|
|
if (mAscendingOrder == true && cmpEleFun(s->data, ele) <= 0 && cmpEleFun(ele, t->data) <= 0) {
|
|
// asc: s <= e <= t
|
|
break;
|
|
} else if(mAscendingOrder == false && cmpEleFun(s->data, ele) >= 0 && cmpEleFun(ele, t->data) >= 0) {
|
|
// desc: s >= e >= t
|
|
break;
|
|
}
|
|
}
|
|
// actually insert
|
|
s->next = newEle;
|
|
newEle->next = t;
|
|
if (t == NULL)
|
|
mUsedLastElement = newEle;
|
|
} else if(asc_first_el != 0 || desc_first_el != 0) {
|
|
// add at front
|
|
newEle->next = mUsedFirstElement;
|
|
mUsedFirstElement = newEle;
|
|
}
|
|
}
|
|
mUsedElementCnts++;
|
|
//printf("_pushElement,%p,data:%p,%d\n",mUsedLastElement,mUsedLastElement->data,mUsedElementCnts);
|
|
// notify only one waiting thread, so that we don't have to check and fall to sleep because we were to slow
|
|
pthread_cond_signal(&mCondGet);
|
|
|
|
return Q_OK;
|
|
}
|
|
|
|
int32_t Queue::_popElement(void **e, bool isWait, int (*cmp)(void *, void *), void *cmpEle)
|
|
{
|
|
// are elements in the queue?
|
|
if (mUsedElementCnts == 0) {
|
|
if (isWait == false) {
|
|
*e = NULL;
|
|
return Q_ERR_NUM_ELEMENTS;
|
|
} else {
|
|
while (mUsedElementCnts == 0 && mAllowedNewData) {
|
|
pthread_cond_wait(&mCondGet, &mMutex);
|
|
}
|
|
if (mUsedElementCnts == 0 && mAllowedNewData == false) {
|
|
return Q_ERR_NONEWDATA;
|
|
}
|
|
}
|
|
}
|
|
|
|
// get first element (which fulfills the requirements)
|
|
Element *elePrev = NULL, *ele = mUsedFirstElement;
|
|
while (cmp != NULL && ele != NULL && 0 != cmp(ele, cmpEle)) {
|
|
elePrev = ele;
|
|
ele = ele->next;
|
|
}
|
|
if (ele != NULL && elePrev == NULL) {
|
|
*e = mUsedFirstElement->data;
|
|
//element is at first, remove this node
|
|
mUsedFirstElement = mUsedFirstElement->next;
|
|
--mUsedElementCnts;
|
|
if (mUsedFirstElement == NULL) {
|
|
mUsedLastElement = NULL;
|
|
}
|
|
//printf("_popElement,%p,data:%p,%p\n",ele,*e,mUsedFirstElement);
|
|
free(ele);
|
|
} else if (ele != NULL && elePrev != NULL) {
|
|
// element is in the middle,remove this node
|
|
elePrev->next = ele->next;
|
|
--mUsedElementCnts;
|
|
*e = ele->data;
|
|
free(ele);
|
|
} else {
|
|
// element is invalid
|
|
*e = NULL;
|
|
return Q_ERR_INVALID_ELEMENT;
|
|
}
|
|
|
|
// notify only one waiting thread
|
|
pthread_cond_signal(&mCondPut);
|
|
|
|
return Q_OK;
|
|
}
|
|
|
|
int32_t Queue::_peekElement(void **e, int32_t pos)
|
|
{
|
|
Element *ele = mUsedFirstElement;
|
|
int32_t elePos = 0;
|
|
if (mUsedElementCnts == 0) {
|
|
return Q_ERR_INVALID_ELEMENT;
|
|
}
|
|
|
|
while (ele != NULL && elePos++ < pos) {
|
|
ele = ele->next;
|
|
}
|
|
|
|
if (ele == NULL) {
|
|
return Q_ERR_INVALID_ELEMENT;
|
|
}
|
|
*e = ele->data;
|
|
return Q_OK;
|
|
}
|
|
|
|
}
|