CAQueue Class Reference

This is a simple FIFO-Queue. More...

#include <CAQueue.hpp>

Collaboration diagram for CAQueue:

Public Member Functions
	CAQueue (UINT32 expectedElementSize=0)
	Give the size of the amount of data what you will add in one step. More...
	~CAQueue ()
	Deletes this Queue and all stored data. More...
SINT32	add (const void *buff, UINT32 size)
	Adds data to the Queue. More...
SINT32	close ()
	Closes the Queue (for writing). More...
SINT32	get (UINT8 *pbuff, UINT32 *psize)
	Gets up to psize number of bytes from the Queue. More...
SINT32	getOrWait (UINT8 *pbuff, UINT32 *psize)
	Gets data from the Queue or waits until some data is available, if the Queue is empty. More...
SINT32	getOrWait (UINT8 *pbuff, UINT32 *psize, UINT32 msTimeOut)
	Gets data from the Queue or waits until some data is available, if the Queue is empty or a timeout is reached. More...
SINT32	peek (UINT8 *pbuff, UINT32 *psize)
	Peeks data from the Queue. More...
SINT32	remove (UINT32 *psize)
	Removes data from the Queue. More...
SINT32	clean ()
	Removes any stored data from the Queue. More...
UINT32	getSize ()
	Returns the size of stored data in byte. More...
UINT32	getSizeLookFree ()
	Returns the size of stored data in byte. More...
bool	isEmpty ()
	Returns true, if the Queue is empty. More...
bool	isClosed ()
	Returns true, if the Queue is closed. More...

Static Public Member Functions
static SINT32	test ()
	Method to test the Queue. More...

Private Attributes
volatile QUEUE *	m_Queue
volatile QUEUE *	m_lastElem
volatile UINT32	m_nQueueSize
volatile bool	m_bClosed
UINT32	m_nExpectedElementSize
CAMutex *	m_pcsQueue
CAConditionVariable *	m_pconvarSize

Detailed Description

This is a simple FIFO-Queue.

You can add data and get them back. This class is thread safe. TODO: The handling of getAndWait is not correct because remove could intercept.... Maybe we do not neeed an other Mutex other than then ConVar....

Definition at line 49 of file CAQueue.hpp.

Constructor & Destructor Documentation

CAQueue()

CAQueue::CAQueue ( UINT32  expectedElementSize = 0 )

inline

Give the size of the amount of data what you will add in one step.

Used for optimizations. Use expectedElementSize=0, if you have no idea about the typicall amount of data added in one call to add().

Definition at line 57 of file CAQueue.hpp.

        {
          m_Queue=NULL;
          m_lastElem=NULL;
          m_nExpectedElementSize=expectedElementSize;
          m_nQueueSize=0;
          m_pcsQueue=new CAMutex();
          m_pconvarSize=new CAConditionVariable();
          m_bClosed=false;
#ifdef QUEUE_SIZE_LOG
          m_nLogSize=0;
#endif
          //m_pHeap=NULL;
          //incHeap();
        }

References m_bClosed, m_lastElem, m_nExpectedElementSize, m_nQueueSize, m_pconvarSize, m_pcsQueue, and m_Queue.

Referenced by test().

Here is the caller graph for this function:

~CAQueue()

CAQueue::~CAQueue

(

)

Deletes this Queue and all stored data.

Definition at line 36 of file CAQueue.cpp.

  {
    clean();
    delete m_pcsQueue;
    m_pcsQueue = NULL;
    delete m_pconvarSize;
    m_pconvarSize = NULL;
  }

References clean(), m_pconvarSize, and m_pcsQueue.

Here is the call graph for this function:

Member Function Documentation

add()

SINT32 CAQueue::add	(	const void *	buff,
		UINT32	size
	)

Adds data to the Queue.

Parameters

buff	pointer to the data buffer
size	size of data to add

Return values

E_UNKNOWN	in case of an error
E_SUCCESS	if succesful

Definition at line 76 of file CAQueue.cpp.

  {
    if(size==0)
      return E_SUCCESS;
    if(buff==NULL)
      return E_UNKNOWN;
    if (m_bClosed)
      return E_UNKNOWN;
    QUEUE* newEntry = new QUEUE;
    newEntry->pBuff=new UINT8[size];
    newEntry->next=NULL;
    newEntry->size=size;
    newEntry->index=0;
    memcpy(newEntry->pBuff,buff,size);
    m_pcsQueue->lock();
    //if(m_pHeap==NULL)
    //  incHeap();
#ifdef _DEBUG
//    if(m_nExpectedElementSize>0&&size>(m_nExpectedElementSize<<1))
    /*if(size>1500)
      {
        CAMsg::printMsg(LOG_DEBUG,"CAQueue::add() WARNING: request for add %u bytes in a queue with expected element size of %u bytes !\n",size,m_nExpectedElementSize);
      }*/
#endif
    if(m_Queue==NULL)
      {
        /*m_Queue=m_pHeap;
        m_pHeap=m_pHeap->next;
        if(size>m_nExpectedElementSize)
          {
            delete[] m_Queue->pBuff;
            m_Queue->pBuff=new UINT8[size];
          }*/
        m_Queue=newEntry;
        m_lastElem=m_Queue;
      }
    else
      {
/*        m_lastElem->next=m_pHeap;
        m_lastElem=m_pHeap;
        m_pHeap=m_pHeap->next;
        if(size>m_nExpectedElementSize)
          {
            delete[] m_lastElem->pBuff;
            m_lastElem->pBuff=new UINT8[size];
          }*/
        m_lastElem->next=newEntry;
        m_lastElem=newEntry;
      }
    m_nQueueSize+=size;
#ifdef QUEUE_SIZE_LOG
    if(m_nLogSize!=0 && m_nQueueSize>m_nLogSize)
      {
        CAMsg::printMsg(LOG_DEBUG,"CAQueue::add() WARNING: queue size is now %u bytes which is above the expected maximum size of %u\n !\n",m_nQueueSize,m_nLogSize);
      }
#endif
    m_pcsQueue->unlock();
    m_pconvarSize->lock();
    m_pconvarSize->signal();
    m_pconvarSize->unlock();
    return E_SUCCESS;
  }

References E_SUCCESS, E_UNKNOWN, _t_queue::index, CAMutex::lock(), m_bClosed, m_lastElem, m_nQueueSize, m_pconvarSize, m_pcsQueue, m_Queue, _t_queue::next, _t_queue::pBuff, CAMsg::printMsg(), CAConditionVariable::signal(), _t_queue::size, and CAMutex::unlock().

Referenced by CAChain::addDataToUpstreamQueue(), CAFirstMix::clean(), CALastMix::clean(), CAFirstMixA::closeConnection(), MemFormatTarget::dumpMem(), CAFirstMixA::loop(), CALastMixA::loop(), CAFirstMixB::loop(), CALastMixB::loop(), CAFirstMixA::notifyAllUserChannels(), producer(), CAMiddleMix::putMixPacketIntoQueueSendToMixBefore(), CAControlChannelDispatcher::sendMessages(), test(), and MemFormatTarget::writeChars().

Here is the call graph for this function:

Here is the caller graph for this function:

clean()

SINT32 CAQueue::clean

(

)

Removes any stored data from the Queue.

Definition at line 47 of file CAQueue.cpp.

  {
    m_pcsQueue->lock();
    while(m_Queue!=NULL)
      {
        delete[] m_Queue->pBuff;
        m_Queue->pBuff = NULL;
        m_lastElem=m_Queue;
        m_Queue=m_Queue->next;
        delete m_lastElem;
      }
/*    while(m_pHeap!=NULL)
      {
        delete[] m_pHeap->pBuff;
        m_lastElem=m_pHeap;
        m_pHeap=m_pHeap->next;
        delete m_lastElem;
      }*/
    m_nQueueSize=0;
    m_lastElem=NULL;
    m_pcsQueue->unlock();
    return E_SUCCESS;
  }

References E_SUCCESS, CAMutex::lock(), m_lastElem, m_nQueueSize, m_pcsQueue, m_Queue, _t_queue::next, _t_queue::pBuff, and CAMutex::unlock().

Referenced by CAFirstMixA::checkUserConnections(), CAChain::closeChainInternal(), CAChain::~CAChain(), and ~CAQueue().

Here is the call graph for this function:

Here is the caller graph for this function:

close()

SINT32 CAQueue::close ( )

inline

Closes the Queue (for writing).

One can still read the remaing bytes out of the queue.

Definition at line 77 of file CAQueue.hpp.

        {
          m_pcsQueue->lock();
          m_bClosed=true;
          m_pcsQueue->unlock();
          if (m_Queue == NULL) //signal possible reader that the queue ist closed...
            {
              m_pconvarSize->lock();
              m_pconvarSize->signal();
              m_pconvarSize->unlock();
            }
          return E_SUCCESS;
        }

References E_SUCCESS, CAMutex::lock(), m_bClosed, m_pconvarSize, m_pcsQueue, m_Queue, CAConditionVariable::signal(), and CAMutex::unlock().

Referenced by CALastMixA::loop().

Here is the call graph for this function:

Here is the caller graph for this function:

get()

SINT32 CAQueue::get	(	UINT8 *	pbuff,
		UINT32 *	psize
	)

Gets up to psize number of bytes from the Queue.

The data is removed from the Queue.

Parameters

pbuff	pointer to a buffer, there the data should be stored
psize	on call contains the size of pbuff, on return contains the size of returned data

Return values

E_SUCCESS	if succesful
E_CLOSED	if the queue is empty and closed
E_UNKNOWN	in case of an error

Definition at line 148 of file CAQueue.cpp.

  {
    if(pbuff==NULL||psize==NULL)
      return E_UNKNOWN;
    if(*psize==0)
      return E_SUCCESS;
    if(m_Queue==NULL)
      {
        *psize=0;
        if ((m_bClosed))
          {
            return E_CLOSED;
          }
        return E_SUCCESS;
      }
    m_pcsQueue->lock();
    UINT32 space=*psize;
    *psize=0;
    while(space>=m_Queue->size)
      {
        memcpy(pbuff,m_Queue->pBuff+m_Queue->index,m_Queue->size);
        *psize+=m_Queue->size;
        pbuff+=m_Queue->size;
        space-=m_Queue->size;
        m_nQueueSize-=m_Queue->size;
        QUEUE* tmp=(QUEUE*)m_Queue;
        m_Queue=m_Queue->next;
        //tmp->next=m_pHeap;
        //m_pHeap=tmp;
        delete[] tmp->pBuff;
        tmp->pBuff = NULL;
        delete tmp;
        if(m_Queue==NULL)
          {
            m_pcsQueue->unlock();
            return E_SUCCESS;
          }
      }
    if(space>0)
      {
        memcpy(pbuff,m_Queue->pBuff+m_Queue->index,space);
        *psize+=space;
        m_Queue->size-=space;
        m_Queue->index+=space;
        m_nQueueSize-=space;
        //memmove(m_Queue->pBuff,m_Queue->pBuff+space,m_Queue->size);
      }
    m_pcsQueue->unlock();
    return E_SUCCESS;
  }

References E_CLOSED, E_SUCCESS, E_UNKNOWN, _t_queue::index, CAMutex::lock(), m_bClosed, m_nQueueSize, m_pcsQueue, m_Queue, _t_queue::next, _t_queue::pBuff, _t_queue::size, and CAMutex::unlock().

Referenced by CAFirstMix::doUserLogin_internal(), getOrWait(), CAFirstMixA::loop(), CALastMixA::loop(), CAFirstMixB::loop(), CALastMixB::loop(), CAFirstMixA::sendToUsers(), and test().

Here is the call graph for this function:

Here is the caller graph for this function:

getOrWait() [1/2]

SINT32 CAQueue::getOrWait	(	UINT8 *	pbuff,
		UINT32 *	psize
	)

Gets data from the Queue or waits until some data is available, if the Queue is empty.

The data is removed from the Queue.

Parameters

pbuff	pointer to a buffer, there the data should be stored
psize	on call contains the size of pbuff, on return contains the size of returned data

Return values

E_SUCCESS	if succesful
E_CLOSED	if the queue is empty and closed
E_UNKNOWN	in case of an error

Definition at line 209 of file CAQueue.cpp.

  {
    m_pconvarSize->lock();
    while (m_Queue == NULL&&!m_bClosed)
      {
        m_pconvarSize->wait();
      }
    SINT32 ret=get(pbuff,psize);
    m_pconvarSize->unlock();
    return ret;
  }

References get(), CAMutex::lock(), m_bClosed, m_pconvarSize, m_Queue, CAMutex::unlock(), and CAConditionVariable::wait().

Referenced by consumer().

Here is the call graph for this function:

Here is the caller graph for this function:

getOrWait() [2/2]

SINT32 CAQueue::getOrWait	(	UINT8 *	pbuff,
		UINT32 *	psize,
		UINT32	msTimeout
	)

Gets data from the Queue or waits until some data is available, if the Queue is empty or a timeout is reached.

The data is removed from the Queue.

Parameters

pbuff	pointer to a buffer, there the data should be stored
psize	on call contains the size of pbuff, on return contains the size of returned data
msTimeout	timeout in milli seconds

Return values

E_SUCCESS	if succesful
E_TIMEDOUT	if timeout was reached
E_UNKNOWN	in case of an error

Definition at line 232 of file CAQueue.cpp.

  {
    m_pconvarSize->lock();
    SINT32 ret;
    while(m_Queue==NULL)
      {
        ret=m_pconvarSize->wait(msTimeout);
        if(ret==E_TIMEDOUT)
          {
            m_pconvarSize->unlock();
            return E_TIMEDOUT;
          }
      }   
    ret=get(pbuff,psize);
    m_pconvarSize->unlock();
    return ret;
  }

References E_TIMEDOUT, get(), CAMutex::lock(), m_pconvarSize, m_Queue, CAMutex::unlock(), and CAConditionVariable::wait().

Here is the call graph for this function:

getSize()

UINT32 CAQueue::getSize ( )

inline

Returns the size of stored data in byte.

Returns

size of Queue

Definition at line 101 of file CAQueue.hpp.

        {
          if (m_pcsQueue)
          {
            m_pcsQueue->lock();
            UINT32 s=m_nQueueSize;
            m_pcsQueue->unlock();
            return s;
          }
          return 0;
        }

References CAMutex::lock(), m_nQueueSize, m_pcsQueue, and CAMutex::unlock().

Referenced by CAFirstMixA::checkUserConnections(), CAFirstMix::doUserLogin_internal(), MemFormatTarget::dumpMem(), CAFirstMixA::loop(), CALastMixA::loop(), CAFirstMixB::loop(), CALastMixB::loop(), CAFirstMixA::sendToUsers(), and test().

Here is the call graph for this function:

Here is the caller graph for this function:

getSizeLookFree()

UINT32 CAQueue::getSizeLookFree ( )

inline

Returns the size of stored data in byte.

This is the look free version which might return a slighty wrong result due to concurrent changes in the queue.

Returns

size of Queue

Definition at line 116 of file CAQueue.hpp.

        {
          return m_nQueueSize;
        }

References m_nQueueSize.

Referenced by CALastMixA::loop().

Here is the caller graph for this function:

isClosed()

bool CAQueue::isClosed ( )

inline

Returns true, if the Queue is closed.

Return values

true,if	Queue is closed
false,otherwise

Definition at line 134 of file CAQueue.hpp.

        {
          return m_bClosed;
        }

References m_bClosed.

Referenced by CALastMixA::loop().

Here is the caller graph for this function:

isEmpty()

bool CAQueue::isEmpty ( )

inline

Returns true, if the Queue is empty.

Return values

true,if	Queue is empty
false,if	Queue contains data

Definition at line 125 of file CAQueue.hpp.

        {
          return (m_Queue==NULL);
        }

References m_Queue.

Referenced by CALastMixA::loop(), CAChain::sendUpstreamData(), and test().

Here is the caller graph for this function:

peek()

SINT32 CAQueue::peek	(	UINT8 *	pbuff,
		UINT32 *	psize
	)

Peeks data from the Queue.

The data is NOT removed from the Queue.

Parameters

pbuff	pointer to a buffer, where the data should be stored
psize	on call contains the size of pbuff, on return contains the size of returned data

Return values

E_SUCCESS	if succesful
E_CLOSED	if the queue is already empty AND closed
E_UNKNOWN	in case of an error

Definition at line 258 of file CAQueue.cpp.

  {
    if(pbuff==NULL||psize==NULL)
      return E_UNKNOWN;
    if(*psize==0)
      return E_SUCCESS;
    m_pcsQueue->lock();
    UINT32 space=*psize;
    *psize=0;
    if(m_Queue==NULL)
      {
        SINT32 ret=E_SUCCESS;
        if(m_bClosed)
          ret=E_CLOSED;
        m_pcsQueue->unlock();
        return ret;
      }
    QUEUE* tmpQueue=(QUEUE*)m_Queue;
    while(space>=tmpQueue->size)
      {
        memcpy(pbuff,tmpQueue->pBuff+tmpQueue->index,tmpQueue->size);
        *psize+=tmpQueue->size;
        pbuff+=tmpQueue->size;
        space-=tmpQueue->size;
        tmpQueue=tmpQueue->next;
        if(tmpQueue==NULL)
          {
            m_pcsQueue->unlock();
            return E_SUCCESS;
          }
      }
    memcpy(pbuff,tmpQueue->pBuff+tmpQueue->index,space);
    *psize+=space;
    m_pcsQueue->unlock();
    return E_SUCCESS;
  } 

References E_CLOSED, E_SUCCESS, E_UNKNOWN, _t_queue::index, CAMutex::lock(), m_bClosed, m_pcsQueue, m_Queue, _t_queue::next, _t_queue::pBuff, _t_queue::size, and CAMutex::unlock().

Referenced by MemFormatTarget::dumpMem(), CALastMixA::loop(), and CAChain::sendUpstreamDataInternal().

Here is the call graph for this function:

Here is the caller graph for this function:

remove()

SINT32 CAQueue::remove

(

UINT32 *

psize

)

Removes data from the Queue.

Parameters

psize

on call contains the size of data to remove, on return contains the size of removed data

Return values

E_SUCCESS	if succesful
E_UNKNOWN	in case of an error

Definition at line 302 of file CAQueue.cpp.

  {
    if(m_Queue==NULL||psize==NULL)
      return E_UNKNOWN;
    if(*psize==0)
      return E_SUCCESS;
    m_pcsQueue->lock();
    UINT32 space=*psize;
    *psize=0;
    while(space>=m_Queue->size)
      {
        *psize+=m_Queue->size;
        space-=m_Queue->size;
        m_nQueueSize-=m_Queue->size;
        QUEUE* tmp=(QUEUE*)m_Queue;
        m_Queue=m_Queue->next;
//        tmp->next=m_pHeap;
//        m_pHeap=tmp;
        delete[] tmp->pBuff;
        tmp->pBuff = NULL;
        delete tmp;
        if(m_Queue==NULL)
          {
            m_pcsQueue->unlock();
            return E_SUCCESS;
          }
      }
    if(space>0)
      {
        *psize+=space;
        m_Queue->size-=space;
        m_nQueueSize-=space;
        m_Queue->index+=space;
        //memmove(m_Queue->pBuff,m_Queue->pBuff+space,m_Queue->size);
      }
    m_pcsQueue->unlock();
    return E_SUCCESS;
  }

References E_SUCCESS, E_UNKNOWN, _t_queue::index, CAMutex::lock(), m_nQueueSize, m_pcsQueue, m_Queue, _t_queue::next, _t_queue::pBuff, _t_queue::size, and CAMutex::unlock().

Referenced by CALastMixA::loop(), and CAChain::sendUpstreamDataInternal().

Here is the call graph for this function:

Here is the caller graph for this function:

test()

SINT32 CAQueue::test ( )

static

Method to test the Queue.

Return values

E_SUCCESS,if

Queue implementation seams to be ok

Definition at line 437 of file CAQueue.cpp.

  {
    CAQueue* pQueue=new CAQueue(1000);
    #define TEST_SIZE 1000000
    UINT8* source=new UINT8[TEST_SIZE];
    UINT8* target=new UINT8[TEST_SIZE];
    getRandom(source,TEST_SIZE);
    UINT32 count=0;
    UINT32 aktSize;
    
    srand((unsigned)time( NULL ));
    
    //Single Thread.....
    //adding
    
    while(TEST_SIZE-count>10000)
        {
          aktSize=rand();
          aktSize%=0xFFFF;
          aktSize%=(TEST_SIZE-count);
          if(pQueue->add(source+count,aktSize)!=E_SUCCESS)
            return E_UNKNOWN;
          count+=aktSize;
          if(pQueue->getSize()!=count)
            return E_UNKNOWN;
        }
    if(pQueue->add(source+count,TEST_SIZE-count)!=E_SUCCESS)
      return E_UNKNOWN;
    if(pQueue->getSize()!=TEST_SIZE)
      return E_UNKNOWN;
    
    //getting
    count=0;
    while(!pQueue->isEmpty())
      {
        aktSize=rand();
        aktSize%=0xFFFF;
        if(pQueue->get(target+count,&aktSize)!=E_SUCCESS)
          return E_UNKNOWN;
        count+=aktSize;
      }
    if(count!=TEST_SIZE)
      return E_UNKNOWN;
    if(memcmp(source,target,TEST_SIZE)!=0)
      return E_UNKNOWN;
    
    //Multiple Threads....
    CAThread* pthreadProducer=new CAThread((UINT8*)"Queue Producer Thread");
    CAThread* pthreadConsumer=new CAThread((UINT8*)"Queue Consumer Thread");
    pthreadProducer->setMainLoop(producer);
    pthreadConsumer->setMainLoop(consumer);
    struct __queue_test t1,t2;
    t1.buff=source;
    t2.buff=target;
    t2.len=t1.len=TEST_SIZE;
    t2.pQueue=t1.pQueue=pQueue;
    pthreadProducer->start(&t1);
    pthreadConsumer->start(&t2);
    pthreadProducer->join();
    pthreadConsumer->join();
    delete pthreadProducer;
    pthreadProducer = NULL;
    delete pthreadConsumer;
    pthreadConsumer = NULL;
    delete pQueue;
    pQueue = NULL;
    if(memcmp(source,target,TEST_SIZE)!=0)
      return E_UNKNOWN;
    
    delete []source;
    source = NULL;
    delete []target;
    target = NULL;
    return E_SUCCESS;
  }

References add(), __queue_test::buff, CAQueue(), consumer(), E_SUCCESS, E_UNKNOWN, get(), getRandom(), getSize(), isEmpty(), CAThread::join(), __queue_test::len, __queue_test::pQueue, producer(), CAThread::setMainLoop(), CAThread::start(), and TEST_SIZE.

Here is the call graph for this function:

Member Data Documentation

m_bClosed

volatile bool CAQueue::m_bClosed

private

Definition at line 155 of file CAQueue.hpp.

Referenced by add(), CAQueue(), close(), get(), getOrWait(), isClosed(), and peek().

m_lastElem

volatile QUEUE* CAQueue::m_lastElem

private

Definition at line 153 of file CAQueue.hpp.

Referenced by add(), CAQueue(), and clean().

m_nExpectedElementSize

UINT32 CAQueue::m_nExpectedElementSize

private

Definition at line 156 of file CAQueue.hpp.

Referenced by CAQueue().

m_nQueueSize

volatile UINT32 CAQueue::m_nQueueSize

private

Definition at line 154 of file CAQueue.hpp.

Referenced by add(), CAQueue(), clean(), get(), getSize(), getSizeLookFree(), and remove().

m_pconvarSize

CAConditionVariable* CAQueue::m_pconvarSize

private

Definition at line 159 of file CAQueue.hpp.

Referenced by add(), CAQueue(), close(), getOrWait(), and ~CAQueue().

m_pcsQueue

CAMutex* CAQueue::m_pcsQueue

private

Definition at line 158 of file CAQueue.hpp.

Referenced by add(), CAQueue(), clean(), close(), get(), getSize(), peek(), remove(), and ~CAQueue().

m_Queue

volatile QUEUE* CAQueue::m_Queue

private

Definition at line 152 of file CAQueue.hpp.

Referenced by add(), CAQueue(), clean(), close(), get(), getOrWait(), isEmpty(), peek(), and remove().

---

The documentation for this class was generated from the following files:

• CAQueue.hpp
• CAQueue.cpp