rvqueue.c

基于h323协议的软phone

/***********************************************************************
Filename   : rvqueue.c
Description: queue functions
************************************************************************
      Copyright (c) 2001,2002 RADVISION Inc. and RADVISION Ltd.
************************************************************************
NOTICE:
This document contains information that is confidential and proprietary
to RADVISION Inc. and RADVISION Ltd.. No part of this document may be
reproduced in any form whatsoever without written prior approval by
RADVISION Inc. or RADVISION Ltd..

RADVISION Inc. and RADVISION Ltd. reserve the right to revise this
publication and make changes without obligation to notify any person of
such revisions or changes.
***********************************************************************/
#include "rvqueue.h"
#include "string.h"

/* Lets make error codes a little easier to type */
#define RvQueueErrorCode(_e) RvErrorCode(RV_ERROR_LIBCODE_CBASE, RV_CBASE_MODULE_QUEUE, (_e))

RvStatus RvQueueInit(void)
{
	return RV_OK;
}

RvStatus RvQueueEnd(void)
{
	return RV_OK;
}

/* Memory is allocated from region bufmem unless the OS allocates its own memory */
/* and doesn't provide any control over it. */
RVCOREAPI RvStatus RVCALLCONV RvQueueConstruct(RvQueue *queue, RvSize_t numitems, RvSize_t itemsize, RvMemory *bufmem)
{
	RvStatus result;
	
#if defined(RV_NULLCHECK)
	if(queue == NULL)
		return RvQueueErrorCode(RV_ERROR_NULLPTR);
#endif
#if defined(RV_RANGECHECK)
	if((numitems == 0) || (itemsize == 0))
		return RvQueueErrorCode(RV_ERROR_OUTOFRANGE);
#endif

	result = RvLockConstruct(&queue->lock);
	if(result != RV_OK)
		return result;

	result = RvSemaphoreConstruct(&queue->emptysem, RvUint32Const(0));
	if(result != RV_OK) {
		RvLockDestruct(&queue->lock);
		return result;
	}
	result = RvSemaphoreConstruct(&queue->fullsem, RvUint32Const(0));
	if(result != RV_OK) {
		RvSemaphoreDestruct(&queue->emptysem);
		RvLockDestruct(&queue->lock);
		return result;
	}
	queue->waitempty = 0;
	queue->waitfull = 0;
	queue->notifyempty = 0;
	queue->notifyfull = 0;
	queue->stopped = RV_FALSE;

	/* Allocate it from the requested region */
	result = RvMemoryAlloc(bufmem, &queue->bufstart, (numitems * itemsize));
	if(result != RV_OK) {
		RvLockDestruct(&queue->lock);
		RvSemaphoreDestruct(&queue->fullsem);
		RvSemaphoreDestruct(&queue->emptysem);
		return result;
	}

	/* Remember last item in buffer for speedier check. */
	queue->bufend = (void *)((RvInt8 *)queue->bufstart + ((numitems - 1) * itemsize));

	queue->firstitem = NULL;
	queue->lastitem = NULL;
	queue->size = numitems;
	queue->itemsize = itemsize;
	queue->curitems = 0;
	return RV_OK;
}

RVCOREAPI RvStatus RVCALLCONV RvQueueDestruct(RvQueue *queue)
{
#if defined(RV_NULLCHECK)
	if(queue == NULL)
		return RvQueueErrorCode(RV_ERROR_NULLPTR);
#endif

	RvLockGet(&queue->lock);
	RvMemoryFree(queue->bufstart);
	RvSemaphoreDestruct(&queue->fullsem);
	RvSemaphoreDestruct(&queue->emptysem);
	RvLockDestruct(&queue->lock);
	return RV_OK;
}

RVCOREAPI RvStatus RVCALLCONV RvQueueReceive(RvQueue *queue, void *buf, RvSize_t bufsize, RvBool wait)
{
	RvSize_t copysize;
	
#if defined(RV_NULLCHECK)
	if((queue == NULL) || (buf == NULL))
		return RvQueueErrorCode(RV_ERROR_NULLPTR);
#endif

	RvLockGet(&queue->lock);
	
	if(queue->waitempty >= queue->curitems) {
		/* queue is empty or other tasks already waiting for available data */
		if(queue->stopped == RV_TRUE) {
		/* Just indicate when queue has been stopped. */
			RvLockRelease(&queue->lock);
			return RvQueueErrorCode(RV_QUEUE_ERROR_STOPPED);
		}
		if(wait == RV_QUEUE_NOWAIT) {
			/* Don't wait, just return and indicate empty */
			RvLockRelease(&queue->lock);
			return RvQueueErrorCode(RV_QUEUE_ERROR_EMPTY);
		}

		/* Wait until something is available. Loop in case queue is cleared */
		/* after semaphore has notified us to get data. */
		do {
			queue->waitempty += 1;
			RvLockRelease(&queue->lock);
			RvSemaphoreWait(&queue->emptysem);
			RvLockGet(&queue->lock);
			queue->waitempty -= 1;
			queue->notifyempty -= 1; /* notify received */

			/* The queue may have stopped while we were waiting. */
			if(queue->stopped == RV_TRUE) {
				/* Just indicate when queue has been stopped. */
				RvLockRelease(&queue->lock);
				return RvQueueErrorCode(RV_QUEUE_ERROR_STOPPED);
			}
		} while(queue->curitems == 0);
	}

	/* Copy out data (safely) */
	copysize = RvMin(queue->itemsize, bufsize);
	memcpy(buf, queue->firstitem, copysize);
	
	/* Adjust queue information */
	queue->curitems -= 1;
	if(queue->firstitem == queue->bufend) {
		queue->firstitem = queue->bufstart; /* Wrap to top of buffer */
	} else queue->firstitem = (void *)((RvInt8 *)queue->firstitem + queue->itemsize);

	/* Wake up someone waiting to put item on queue. */
	if(queue->waitfull > queue->notifyfull) {
		queue->notifyfull += 1;
		RvSemaphorePost(&queue->fullsem);
	}

	RvLockRelease(&queue->lock);

	return RV_OK;
}

RVCOREAPI RvStatus RVCALLCONV RvQueueSend(RvQueue *queue, void *buf, RvSize_t bufsize, RvInt priority, RvBool wait)
{
	RvSize_t copysize;
	void *newitem;

#if defined(RV_NULLCHECK)
	if((queue == NULL) || (buf == NULL))
		return RvQueueErrorCode(RV_ERROR_NULLPTR);
#endif

	RvLockGet(&queue->lock);

	if(queue->stopped == RV_TRUE) {
		/* Just indicate when queue has been stopped. */
		RvLockRelease(&queue->lock);
		return RvQueueErrorCode(RV_QUEUE_ERROR_STOPPED);
	}
	
	if(queue->waitfull >= (queue->size - queue->curitems)) {
		/* queue is full or tasks already waiting for available space */
		if(wait == RV_QUEUE_NOWAIT) {
			/* Don't wait, just return and indicate full */
			RvLockRelease(&queue->lock);
			return RvQueueErrorCode(RV_QUEUE_ERROR_FULL);
		}
		
		/* Wait until there is space available. */
		queue->waitfull += 1;
		RvLockRelease(&queue->lock);
		RvSemaphoreWait(&queue->fullsem);
		RvLockGet(&queue->lock);
		queue->waitfull -= 1;
		queue->notifyfull -= 1; /* notify received */

		/* The queue may have stopped while we were waiting. */
		if(queue->stopped == RV_TRUE) {
			/* Just indicate when queue has been stopped. */
			RvLockRelease(&queue->lock);
			return RvQueueErrorCode(RV_QUEUE_ERROR_STOPPED);
		}
	}

	/* Adjust queue information */
	if(queue->curitems == 0) {
		/* queue is empty, putting on front or end is the same */
		queue->firstitem = queue->bufstart;
		queue->lastitem = queue->bufstart;
		newitem = queue->bufstart;
	} else {
		if(priority == RV_QUEUE_SEND_NORMAL) {
			/* Put on end of queue */
			if(queue->lastitem == queue->bufend) {
				queue->lastitem = queue->bufstart; /* Wrap to top of buffer */
			} else queue->lastitem = (void *)((RvInt8 *)queue->lastitem + queue->itemsize);
			newitem = queue->lastitem;
		} else {
			/* Put on front of queue (URGENT) */
			if(queue->firstitem == queue->bufstart) {
				queue->firstitem = queue->bufend; /* Wrap to end of buffer */
			} else queue->firstitem = (void *)((RvInt8 *)queue->firstitem - queue->itemsize);
			newitem = queue->firstitem;
		}
	}
	queue->curitems += 1;

	/* Copy in data (safely) */
	copysize = RvMin(queue->itemsize, bufsize);
	memcpy(newitem, buf, copysize);

	/* Wake up someone waiting to get item from queue */
	if(queue->waitempty > queue->notifyempty) {
		queue->notifyempty += 1;
		RvSemaphorePost(&queue->emptysem);
	}

	RvLockRelease(&queue->lock);

	return RV_OK;
}

/* Stop the queue: nothing new can be put on the queue and */
/* tasks will not block on send or receive. */
RVCOREAPI RvStatus RVCALLCONV RvQueueStop(RvQueue *queue)
{
#if defined(RV_NULLCHECK)
	if(queue == NULL)
		return RvQueueErrorCode(RV_ERROR_NULLPTR);
#endif
	
	RvLockGet(&queue->lock);
	if(queue->stopped == RV_TRUE) {
		RvLockRelease(&queue->lock);
		return RV_OK;
	}
	
	queue->stopped = RV_TRUE;

	/* If any tasks are blocked trying to send, release them. */
	while(queue->notifyfull < queue->waitfull) {
		queue->notifyfull += 1;
		RvSemaphorePost(&queue->fullsem);
	}
	
	/* If any tasks are blocked trying to receive, release them. */
	while(queue->notifyempty < queue->waitempty) {
		queue->notifyempty += 1;
		RvSemaphorePost(&queue->emptysem);