rbufflib.c

VXWORKS源代码

    return (OK);
    }


/*******************************************************************************
*
* wvRBuffMgr - The function is spawned as a task to manage WV rBuff extension.
*
* It waits for a msg which carries the ID of a buffer which has had a change
* of the number of empty buffers held. If there are less the requested to be
* reserved than an attempt will be made to make more available. If there are
* more empty buffers than required than some will be freed.
*
* For each msg received, the task will achieve the required number of free
* buffers. This may result in redundant msgs being received.
*
* This manager is specific to WindView
*
* NOMANUAL
*/

int wvRBuffMgr
    (
    WV_RBUFF_MGR_TYPE * rBuffMgrId  /* id of the WindView rBuffMgr structure */	
    )
    {
    UINT32		    	busy;
    int          	    	key;
    RBUFF_PTR    	    	newBuff;
    WV_RBUFF_MGR_MSG_TYPE* 	pMsg;




#ifdef RBUFF_DEBUG
        logMsg ("rBuffMgr %p created\n",rBuffMgrId,0,0,0,0,0);
#endif

    for(;;)
        {

	semTake (&(rBuffMgrId->msgSem), WAIT_FOREVER);
	pMsg = &(rBuffMgrId->msg[rBuffMgrId->msgReadIndex]);

#ifdef RBUFF_DEBUG
        logMsg ("rBuffMgr: (%d) rBuff %p type 0x%x arg 0x%x\n",
		rBuffMgrId->msgReadIndex,
                pMsg->ringId, 
                pMsg->msgType, 
                pMsg->arg, 
                0, 0);
#endif

        busy = TRUE;

        /* Determine what needs to be done */

        switch (pMsg->msgType)
            {
            case RBUFF_MSG_ADD :
                {

                /*
                 *  This message may be generated one or more times by
                 *  event logging, and the condition that caused the msg
                 *  may no longer be TRUE, so just take this as a prompt
                 *  to see if the ring needs extending.
                 */

                /* Make sure that we don't request extension recursively */

                pMsg->ringId->nestLevel++;

                while
                    (pMsg->ringId->info.emptyBuffs < RBUFF_EMPTY_KEEP &&
                    (pMsg->ringId->info.currBuffs  < pMsg->ringId->info.maxBuffs ||
                        pMsg->ringId->info.options & RBUFF_WRAPAROUND) &&
                    busy)
                    {

                    /* Looks like we are going to need a buffer */

                    newBuff = (RBUFF_PTR)
                        memPartAlloc (pMsg->ringId->info.srcPart,
                            sizeof(RBUFF_BUFF_TYPE) + pMsg->ringId->info.buffSize);

                    /* of course, newBuff may be NULL... */

                    key = intLock();

#ifdef RBUFF_DEBUG
                    logMsg ("rBuffMgr: adding buffer %p\n",newBuff,0,0,0,0,0);
#endif

		    if (rBuffAdd (pMsg->ringId, newBuff) == ERROR)
                        {
                        /* ring cannot be extended so give up trying */

                        busy = FALSE;
                        }

                    intUnlock (key);
                    }
                }

                pMsg->ringId->nestLevel--;

            break;

            case RBUFF_MSG_FREE :
                {
                /* This message carries a ptr to the buffer to be freed */

#ifdef RBUFF_DEBUG
                logMsg ("rBuffMgr: freeing buffer %p\n", pMsg->arg, 0,0,0,0,0);
#endif

                memPartFree (pMsg->ringId->info.srcPart, (UINT8 *) pMsg->arg);
                }
            break;

            case RBUFF_MSG_FULL :
    		logMsg ("Ring Buffer %p full. Stopped event logging.\n",
    		        (unsigned int)pMsg->ringId,0,0,0,0,0);
	    break;

	    default :
                logMsg ("Unknown message type %d\n", pMsg->msgType, 0,0,0,0,0);
	    break;
            }

         if (++rBuffMgrId->msgReadIndex > (WV_RBUFF_MGR_MSGQ_MAX - 1))
             {
             rBuffMgrId->msgReadIndex = 0;
             }

         pMsg->ringId->msgOutstanding = FALSE;

         }
    }

/*******************************************************************************
*
* wvRBuffMgrPrioritySet - set the priority of the WindView rBuff manager (WindView)
*
* This routine changes the priority of the `tWvRBuffMgr' task to the value of 
* <priority>.  Priorities range from 0, the highest priority, to 255, the
* lowest priority.  If the task is not yet running, this priority is
* used when it is spawned.
* 
* RETURNS: OK, or ERROR if the priority can not be set.
*
* SEE ALSO: taskPrioritySet(),
* .pG "Basic OS"
*/

STATUS wvRBuffMgrPrioritySet
    (
    int	     priority           /* new priority */
    )
    {

    if (!rBuffLibInstalled)
        {
    	if (rBuffLibInit() == OK)
	    rBuffLibInstalled = TRUE;
    	else
            return (ERROR);
	}

    if (wvRBuffMgrId->tid != 0)
 	if (taskPrioritySet (wvRBuffMgrId->tid, priority) != OK)
	    return (ERROR);

    wvRBuffMgrId->priorityDefault = priority;
    return (OK);
    }

/*******************************************************************************
*
* rBuffUpload - Copy data that is added to the buffer to the specified 'fd'
*
* NOMANUAL
*/

STATUS rBuffUpload
    (
    BUFFER_ID 	buffId,		/* generic identifier for this buffer */
    int 	fd
    )
    {
    RBUFF_ID 	rBuff;		/* specific identifier for this rBuff */
    UINT8 *	src;
    int 	lockKey;
    STATUS 	result = OK;

    /* Get access to the private members of this particular rBuff. */

    rBuff = (RBUFF_ID) buffId;

    rBuff->fd = fd;

    while (result == OK)
        {

        /* Wait for the dinner gong */

        logMsg ("Waiting for data\n",0,0,0,0,0,0);

        semTake (&rBuff->buffDesc.threshXSem,WAIT_FOREVER);

        logMsg("Uploading...\n",0,0,0,0,0,0);

        /* tuck in... */

        while (result == OK && rBuff->info.dataContent >
              (rBuff->info.threshold / 4))
            {
            UINT32 bytesToWrite = rBuffReadReserve(buffId, &src);
            UINT32 bytesWritten;

            if (bytesToWrite > 4096)
                {
                    bytesToWrite = 4096;
                }

#ifdef RBUFF_DEBUG

            logMsg ("%#x bytes from %p (dataContent=%#x)\n",
                bytesToWrite,
                (int) src,
                rBuff->info.dataContent,0,0,0);
#endif


            /* move the data */

#if 0
{
UINT8 buffer[100];

bcopy(src,buffer,bytesToWrite);
buffer[bytesToWrite] = '\0';

printf("%s\n",buffer);
}
#endif

            if ((bytesWritten =
                write (fd, src, bytesToWrite)) == -1)
                {
                logMsg ("Upload error!\n",0,0,0,0,0,0);

                if (rBuff->errorHandler)
                    {

                    /* Dial 999 / 911 */

                    (*rBuff->errorHandler)();
                    }

                result = ERROR;
                }
            else
                {
                /* move the buffer pointers along */

                lockKey = intLock();

                result = rBuffReadCommit (buffId,bytesWritten);

                intUnlock (lockKey);

                taskDelay (0);
                }
            }
        }
    return (result);
    }

#else  /* GENERIC_RBUFF */

/*
 * The code below constitutes the generic ring buffer scheme on which
 * WindView 2.0 was originally based. If there is a need to go back to
 * it, #define GENERIC_RBUFF at the top of this file or build with
 * EXTRA_DEFINE="-DGENERIC_RBUFF" in the make command line
 */

#include "vxWorks.h"
#include "semLib.h"
#include "classLib.h"
#include "errno.h"
#include "intLib.h"
#include "logLib.h"
#include "taskLib.h"
#include "fioLib.h"
#include "rBuffLib.h"


#include "private/wvBufferP.h"
#include "private/classLibP.h"
#include "private/objLibP.h"
#include "private/workQLibP.h"

#include "stdio.h"
#include "string.h"
#include "unistd.h"

extern int rBuffMgr();

#if (CPU_FAMILY == I80X86)
extern void portWorkQAdd1();
#endif

/* locals */

#if defined(__STDC__) || defined(__cplusplus)

LOCAL STATUS rBuffAdd
    (
    RBUFF_ID  rBuff,
    RBUFF_PTR buffToAdd
    );

LOCAL STATUS  rBuffFree
    (
    RBUFF_ID  rBuff,
    RBUFF_PTR buffToFree
    );

LOCAL STATUS rBuffHandleEmpty
    (
    RBUFF_ID rBuff
    );

#else	/* __STDC__ */

LOCAL STATUS rBuffAdd();
LOCAL STATUS rBuffFree();
LOCAL STATUS rBuffHandleEmpty();

#endif	/* __STDC__ */

extern  BOOL kernelState;

#if (CPU_FAMILY == I80X86)

#define RBUFF_SEND_MSG(RMSG_TYPE,RMSG_PRI,RMSG_ID,RMSG_PARAM) \
if((!(RMSG_ID)->nestLevel++) && !(RMSG_ID)->msgOutstanding)   \
{                                                             \
    (RMSG_ID)->msgOutstanding = TRUE;                         \
    (RMSG_ID)->msg[(RMSG_ID)->msgWriteIndex][0] = RMSG_TYPE;  \
    (RMSG_ID)->msg[(RMSG_ID)->msgWriteIndex][1] = (UINT32) RMSG_PARAM; \
                                                              \
    if(++(RMSG_ID)->msgWriteIndex > RBUFF_MAX_MSGS)           \
    {                                                         \
        (RMSG_ID)->msgWriteIndex = 0;                         \
    }                                                         \
                                                              \
    if(kernelState)                                           \
    {                                                         \
        portWorkQAdd1((FUNCPTR)semCGiveDefer, (int)&(RMSG_ID)->msgSem); \
    }                                                         \
    else                                                      \
    {                                                         \
        semCGiveDefer(&(RMSG_ID)->msgSem);                    \
    }                                                         \
}                                                             \
                                                              \
(RMSG_ID)->nestLevel--;

#else /* CPU_FAMILY == I80X86 */

#define RBUFF_SEND_MSG(RMSG_TYPE,RMSG_PRI,RMSG_ID,RMSG_PARAM) \
if((!(RMSG_ID)->nestLevel++) && !(RMSG_ID)->msgOutstanding)   \
{                                                             \
    (RMSG_ID)->msgOutstanding = TRUE;                         \
    (RMSG_ID)->msg[(RMSG_ID)->msgWriteIndex][0] = RMSG_TYPE;  \
    (RMSG_ID)->msg[(RMSG_ID)->msgWriteIndex][1] = (UINT32) RMSG_PARAM; \
                                                              \
    if(++(RMSG_ID)->msgWriteIndex > RBUFF_MAX_MSGS)           \
    {                                                         \
        (RMSG_ID)->msgWriteIndex = 0;                         \
    }                                                         \
                                                              \
    if(kernelState)                                           \
    {                                                         \
        workQAdd1((FUNCPTR)semCGiveDefer, (int)&(RMSG_ID)->msgSem); \
    }                                                         \
    else                                                      \
    {                                                         \
        semCGiveDefer(&(RMSG_ID)->msgSem);                    \
    }                                                         \
}                                                             \
                                                              \
(RMSG_ID)->nestLevel--;

#endif /* CPU_FAMILY == I80X86 */

LOCAL BOOL   rBuffLibInstalled = FALSE;   /* protect from multiple inits */

/* global variables */

OBJ_CLASS rBuffClass;                   /* rBuff object Class */
CLASS_ID  rBuffClassId = &rBuffClass;   /* rBuff class ID */
int       rBuffMgrPriorityDefault = RBUFF_MGR_PRIORITY; /* default ring buff */
                                        /* manager priority */

/*******************************************************************************
*
* rBuffLibInit - initialise the ring of buffers library
*
* This routine initialises the ring of buffers library. If INCLUDE_RBUFF is
* defined in configAll.h, it is called by the root task, usrRoot(), in
* usrConfig.c.
*
* NOMANUAL
*
* RETURNS: OK, or ERROR if the library could not be initialized.
*/

STATUS rBuffLibInit (void)
    {   

    if(rBuffLibInstalled) {
        return(TRUE);
    }

    rBuffLibInstalled = TRUE;

    /* Initialise the rBuff Manager */

    /* initialize the rBuff class structure */

    return (classInit (rBuffClassId,
                sizeof (RBUFF_TYPE),
                OFFSET (RBUFF_TYPE, buffDesc),
                (FUNCPTR) rBuffCreate,
                (FUNCPTR) NULL,
                (FUNCPTR) rBuffDestroy));
    }
 

/*******************************************************************************
*
* rBuffCreate - create an extendable ring of buffers
*
* This routine creates an extendable ring of buffers from
* the specified partition.
*
* NOMANUAL
*/

BUFFER_ID rBuffCreate
    (
    void *Params
    )
    {

    RBUFF_ID              rBuff;
    UINT32                count;
    RBUFF_PTR             newBuff;
    rBuffCreateParamsType *rBuffCreateParams = Params;

    if ((!rBuffLibInstalled) && (rBuffLibInit()) == OK)
	{
	rBuffLibInstalled = TRUE;
	}

    if(!rBuffLibInstalled)
        {
#ifdef RBUFF_DEBUG
        logMsg ("rBuff: rBuffLib not installed\n",0,0,0,0,0,0);
#endif

        return(NULL);
        }

    /* validate params */

    if (rBuffCreateParams->minimum < 2 ||
            ((rBuffCreateParams->minimum > rBuffCreateParams->maximum) ||
               (rBuffCreateParams->maximum < 2 &&
                rBuffCreateParams->maximum != RBUFF_MAX_AVAILABLE)) ||
            rBuffCreateParams->buffSize == 0)
        {
#ifdef RBUFF_DEBUG
        logMsg ("rBuff: invalid params\n",0,0,0,0,0,0);
#endif