rbufflib.c

VXWORKS源代码

        /* drop the excess buffers */

        excessBuffs = rBuff->info.currBuffs - rBuff->info.minBuffs;

        for (count=0;count < excessBuffs;count++)
            {
            RBUFF_PTR tempPtr;

            tempPtr = rBuff->buffWrite->next; /* should not be NULL! */

            rBuffFree (rBuff,rBuff->buffWrite);

            rBuff->buffWrite = tempPtr;
            }
        }

    /* Make sure the 'read' buffer pointer points somewhere sensible */

    rBuff->buffRead = rBuff->buffWrite;

    /* and reset the byte ptrs */

    rBuff->dataRead         =
    rBuff->dataWrite        = rBuff->buffWrite->dataStart;

    rBuff->info.currBuffs        =
    rBuff->info.maxBuffsActual   = rBuff->info.minBuffs;

    /* now traverse the ring resetting the individual buffers */

    {
    RBUFF_PTR backHome = rBuff->buffWrite;
    RBUFF_PTR currBuff = rBuff->buffWrite;

    do
     	{
        currBuff->dataLen    = 0;
        currBuff->spaceAvail = rBuff->info.buffSize;

        currBuff = currBuff->next;
        }
    while (currBuff != backHome);

    }

    /* Reset msg passing mechanism */

    rBuff->msgOutstanding        = FALSE;

    rBuff->msgWriteIndex         =
    rBuff->msgReadIndex          = 0;

    rBuff->nestLevel = 0;

    /* reset info */

    rBuff->info.emptyBuffs       = rBuff->info.minBuffs;
    rBuff->info.dataContent      =
    rBuff->info.writesSinceReset =
    rBuff->info.readsSinceReset  =
    rBuff->info.timesExtended    =
    rBuff->info.timesXThreshold  =
    rBuff->info.bytesWritten     =
    rBuff->info.bytesRead        = 0;

    semGive (&rBuff->bufferFull);

    RBUFF_UNLOCK (rBuff);

    return (OK);
}


/*******************************************************************************
*
* rBuffNBytes - Returns the total number of bytes held in a ring of buffers
*
* NOMANUAL
*/

INT32 rBuffNBytes
    (
    BUFFER_ID buffId		/* generic identifier for this buffer */
    )
    {
    RBUFF_ID rBuff;		/* specific identifier for this rBuff */

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

    rBuff = (RBUFF_ID) buffId;

    return (rBuff->info.dataContent);
    }


/*******************************************************************************
*
* rBuffAdd - Add a new buffer to an extendable ring of buffers
*
* This routine adds a further buffer to an already created or partially created
* extendable ring of buffers unless the maximum number is already allocated.
*
* This function assumes mutually exclusive access is guaranteed by the caller.
*
* NOMANUAL
*/

STATUS rBuffAdd
    (
    RBUFF_ID 	rBuff,
    RBUFF_PTR 	newBuff
    )
    {
    if (rBuff->info.emptyBuffs >= RBUFF_EMPTY_KEEP &&
        rBuff->info.currBuffs  >= rBuff->info.minBuffs)
        {
        /*
         *  OK, the uploader got in so now we don't need to extend the
         *  buffer.
         */

#ifdef RBUFF_DEBUG
        logMsg ("rBuff: abandoned add\n",0,0,0,0,0,0);
#endif

        if (newBuff)
            {
            memPartFree (rBuff->info.srcPart, (UINT8 *) newBuff);
            }

        return (OK);
        }

    if (rBuff->info.currBuffs == rBuff->info.maxBuffs || newBuff == NULL)
        {

        /*
         *  We are at the maximum ring size already or the source
         *  partition is exhausted, there's still hope...
         *
         *  ...can we wrap-around?
         */

        if (!(rBuff->info.options & RBUFF_WRAPAROUND))
            {
            /* We aren't allowed to wrap around */

            return (ERROR);
            }
        else
            {
            RBUFF_PTR reclaim;

            /*  We can wrap around by comandeering the 'oldest'
             *  buffer in the ring.
             *
             *  It is possible (probable?) that this buffer is the
             *  one to be read from next so handle this.
             */

            reclaim = rBuff->buffWrite->next;

            if (rBuff->buffRead == reclaim)
                {
                /* Move the reader along */

                rBuff->buffRead = rBuff->buffRead->next;
                rBuff->dataRead = rBuff->buffRead->dataStart;
                }

            rBuff->info.dataContent  -= reclaim->dataLen;

            /* Reset the buffer */

            reclaim->dataLen    = 0;
            reclaim->spaceAvail = rBuff->info.buffSize;

            rBuff->info.emptyBuffs++;

            return (OK);
            }
        }

    newBuff->dataStart  = (UINT8 *) newBuff + sizeof(RBUFF_BUFF_TYPE);
    newBuff->dataEnd    = newBuff->dataStart + rBuff->info.buffSize;
    newBuff->spaceAvail = rBuff->info.buffSize;
    newBuff->dataLen    = 0;

    /* Now link the buffer into the ring */

    if (rBuff->buffWrite != NULL)
        {
        RBUFF_PTR tempPtr;

        tempPtr = rBuff->buffWrite->next;

        rBuff->buffWrite->next = newBuff;

        /* close the ring */

        newBuff->next = tempPtr;

        }
    else
        {

        /* ring is empty, must be creation time */

        rBuff->buffWrite =
        rBuff->buffRead  =
        newBuff->next    = newBuff;
        }

    /* Maintain statistics */

    if (++(rBuff->info.currBuffs) > rBuff->info.maxBuffsActual)
        {
        rBuff->info.maxBuffsActual++;
        }

    rBuff->info.timesExtended++;
    rBuff->info.emptyBuffs++;

    return (OK);
}


/*******************************************************************************
*
* rBuffFree - Free a specified buffer
*
* This function must only be called from task level.
* Semaphores are used to protect access at task-level. However, protection
* from access at interrupt level must be guaranteed by the caller.
*
* NOMANUAL
*/

STATUS rBuffFree
    (
    RBUFF_ID 	rBuff,
    RBUFF_PTR 	buffToFree
    )
    {
    RBUFF_PTR 	ptrToBuff, backHome, ptrToPreviousBuff;

    if (buffToFree == NULL)
        {
        return(ERROR);
        }

    /* find the pointer to the buffer to be freed */

    ptrToBuff         = rBuff->buffWrite->next;
    backHome          = rBuff->buffWrite;
    ptrToPreviousBuff = rBuff->buffWrite;

    while ((ptrToBuff != buffToFree) && (ptrToBuff != backHome))
        {
        ptrToPreviousBuff = ptrToBuff;
        ptrToBuff         = buffToFree;
        }

    if (ptrToBuff == backHome)
        {
        /* an error has occured, possibly invalid buff ptr passed */

        return (ERROR);
        }

    /* OK, de-link the buffer to be freed */

    ptrToPreviousBuff->next = buffToFree->next;

    /* Schedule the rBuff Mgr to actually free the buffer */

    RBUFF_SEND_MSG (RBUFF_MSG_FREE, MSG_PRI_NORMAL, rBuff, buffToFree);

    /* Maintain statistics */

    rBuff->info.emptyBuffs--;
    rBuff->info.currBuffs--;

    return (OK);
    }


/*******************************************************************************
*
* rBuffDestroy - Destroy an extendable ring of buffers
*
* This routine destroys an already created or partially created extendable ring
* of buffers. This loses any data held in the buffer. All resources held are
* returned to the system.
*
* NOMANUAL
*/

STATUS rBuffDestroy
    (
    BUFFER_ID 	buffId		/* generic identifier for this buffer */
    )
    {
    RBUFF_ID 	rBuff;		/* specific identifier for this rBuff */
    RBUFF_PTR 	tempPtr;

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

    rBuff = (RBUFF_ID) buffId;

    RBUFF_LOCK (rBuff);

#if 0
    if (OBJ_VERIFY (rBuff, rBuffClassId) != OK)	/* validate rBuff ID */
	{
	((OBJ_ID)rBuff)->pObjClass, rBuffClassId);
	RBUFF_UNLOCK (rBuff);
	return (ERROR);
	}
#endif

    objCoreTerminate (&rBuff->buffDesc.objCore);   /* invalidate this object */

    /* Break the ring and traverse the list, freeing off   */
    /* the buffers. buffWrite is used as the start point   */
    /* this works if the ring creation is abandoned early. */

    if (rBuff->buffWrite != NULL)
        {

        /* Break the ring */

        tempPtr = rBuff->buffWrite->next;
        rBuff->buffWrite->next = NULL;
        rBuff->buffWrite = tempPtr;

        /* traverse and free */

        while(rBuff->buffWrite != NULL)
            {
            tempPtr = rBuff->buffWrite->next;
            memPartFree (rBuff->info.srcPart,(UINT8 *)rBuff->buffWrite);
            rBuff->buffWrite = tempPtr;
            }
        }
    else
        {
        /* ring hasn't been created so don't worry about it */
        }

    /* now free off the control structure */

    semTerminate (&rBuff->buffDesc.threshXSem);
    semTerminate (&rBuff->readBlk);
    semTerminate (&rBuff->bufferFull);

    objFree (rBuffClassId, (UINT8 *) rBuff);

    /* we're done */

    return (OK);
    }


/*******************************************************************************
*
* rBuffSetFd - Set a new fd as the data upload destination
*
*
* NOMANUAL
*/

STATUS rBuffSetFd
    (
    BUFFER_ID 	buffId,		/* generic identifier for this buffer */
    int 	fd
    )
    {
    RBUFF_ID 	rBuff;		/* specific identifier for this rBuff */

    RBUFF_LOCK (rBuff);

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

    rBuff = (RBUFF_ID) buffId;
    rBuff->fd = fd;
    RBUFF_UNLOCK (rBuff);

    return (OK);
    }


/*******************************************************************************
*
* rBuffHandleEmpty - Handles a newly empty buffer by deciding whether to free
*                    it off.
*
* This function assumes mutually exclusive access is guaranteed by the caller.
*
* NOMANUAL
*/

LOCAL STATUS rBuffHandleEmpty
    (
    RBUFF_ID rBuff
    )
    {
    BOOL      moveAlong = TRUE;
    RBUFF_PTR buffToFree;

    /*
     *  Determine if we are to move around the ring and if the
     *  newly empty buffer is to be freed, and then do the actions.
     *  This avoids having to fumble with ptrs as we move around
     *  from a buffer that has been freed.
     */

    if (rBuff->buffRead == rBuff->buffWrite)
        {

        /*
         *  We've caught the writer so might as well reset this
         *  buffer and make efficient use of it.
         */

        rBuff->dataRead  =
            rBuff->dataWrite = rBuff->buffWrite->dataStart;

        rBuff->buffRead->spaceAvail = rBuff->info.buffSize;
        rBuff->buffRead->dataLen     = 0;

        /* In this case, we don't move along to the next buffer */

        moveAlong = FALSE;
        }

    if (++(rBuff->info.emptyBuffs) > RBUFF_EMPTY_KEEP &&
           rBuff->info.currBuffs > rBuff->info.minBuffs)
        {
        /* We already have enough empty buffs, so free one off */

        if (rBuff->buffRead != rBuff->buffWrite)
            {
            buffToFree = rBuff->buffRead;
            }
        else
            {
            /*
             *  As the reader and writer are in the same buffer
             *  we know it's safe to free off the next buffer.
             */

            buffToFree = rBuff->buffRead->next;
            }
        }
    else
        {
        /* we've just emptied one of the 'core' buffers */

        rBuff->buffRead->spaceAvail = rBuff->info.buffSize;
        rBuff->buffRead->dataLen    = 0;

        /* Don't free */

        buffToFree = NULL;
        }

    if (moveAlong)
        {
        /* Move round the ring */

        rBuff->buffRead = rBuff->buffRead->next;
        rBuff->dataRead = rBuff->buffRead->dataStart;
        }
    else
        {
        /* reset flag for next time */

        moveAlong = TRUE;
        }

    if (/* there's a */ buffToFree)
        {
#ifdef RBUFF_DEBUG
        logMsg ("rBuff: Freeing buffer %p\n",
            buffToFree,0,0,0,0,0);
#endif
        rBuffFree (rBuff, buffToFree);
        }