tools/Queue.cpp - yocto/linux/multimedia/libvideorender - Git Browser for ODROID

 /*
  * 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 Tls {

 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;
 }

 }
	/*
	* 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 Tls {

	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;
	}

	}