osapi_message.c

这是DVD中伺服部分的核心代码

    /* Return success */
    return (iRet);

error_handler:

    /* Return failure */
    return (OS_FAILURE);
}

/**
 * OS Message Queue Receive function.
 *
 * @param mqId     - message queue identifier
 * @param *strMsg  - message pointer, the location of where the received message is to be placed after it has been received
 * @param uiSize   - message size (not used)
 * @param iTimeout - send timeout value, in ticks. (-1 = wait forever, 0 = no wait, > 1 = wait)
 *
 * @retval
 *    OS_OK if successful
 *    OS_FAILURE if not successful
 *    OS_TIMEOUT if a timeout occurred
 *
 * @verified
 *    Yes.
 */
#if defined(NDEBUG)
OS_STATUS OS_MsgQReceive(OS_MSG_Q_ID mqId, char *strMsg, ULONG ulSize, int iTimeout)
#else
OS_STATUS OS_MsgQReceiveDbg(OS_MSG_Q_ID mqId, char *strMsg, ULONG ulSize, int iTimeout, char *srcFile, int srcLine)
#endif
{
    /* Check the validity of message queue */
    if (mqId == 0)
    {
        #if defined(NDEBUG)
        #else
            DbgPrint(("%s: BAD POINTER.  Called by %s:%u\n", __FUNCTION__, srcFile, srcLine));
        #endif
        return (OS_FAILURE);
    }

    /* Receive message from the head by default */
    return (OS_MsgQReceiveOption(mqId, strMsg, ulSize, iTimeout, OS_MSG_Q_HEAD) );
}

/**
 * OS Message Queue Remove function.
 *
 * @param mqId     - message queue identifier
 * @param *strMsg  - message pointer, the location of where the removed message is to be placed after it has been removed
 * @param uiSize   - message size (not used)
 *
 * @retval
 *    OS_OK if successful
 *    OS_FAILURE if not successful
 *    OS_TIMEOUT if msg queue is empty
 *
 * @verified
 *    Yes.
 */
#if defined(NDEBUG)
OS_STATUS OS_MsgQRemove(OS_MSG_Q_ID mqId, char *strMsg, ULONG ulSize)
#else
OS_STATUS OS_MsgQRemoveDbg(OS_MSG_Q_ID mqId, char *strMsg, ULONG ulSize, char *srcFile, int srcLine)
#endif
{
    /* Check the validity of message queue */
    if (mqId == 0)
    {
        #if defined(NDEBUG) 
        #else
            DbgPrint(("%s: BAD POINTER.  Called by %s:%u\n", __FUNCTION__, srcFile, srcLine));
        #endif
        return (OS_FAILURE);
    }

    /* Receive message from the head by default */
    return (OS_MsgQReceiveOption(mqId, strMsg, ulSize, OS_NO_WAIT, OS_MSG_Q_TAIL) );
}

/**
 * OS Message Queue Receive Option function. Same as OS_MsgQReceive, but allow message to received
 *      from either the head or the tail of the message queue.
 *
 * @param mqId     - message queue identifier
 * @param *strMsg  - message pointer, the location of where the received message is to be placed after it has been received
 * @param uiSize   - message size (not used)
 * @param iTimeout - send timeout value, in ticks. (-1 = wait forever, 0 = no wait, > 1 = wait)
 * @param iOption  - head or tail of message queue (OS_MSG_Q_HEAD or OS_MSG_Q_TAIL)
 *
 * @retval
 *    OS_OK if successful
 *    OS_FAILURE if not successful
 *    OS_TIMEOUT if a timeout occurred
 *
 * @verified
 *    Yes.
 */
static OS_STATUS OS_MsgQReceiveOption(OS_MSG_Q_ID mqId, char *strMsg, ULONG ulSize, int iTimeout, int iOption)
{
    OS_STATUS iRet  = OS_OK;
    OS_MBOX   *pmsg = (OS_MBOX *)mqId;
    BOOLEAN        waitedOnEmptySemaphore = FALSE;

    /* Check the validity of message queue */
    if (mqId == 0)
    {
        DbgPrint(("%s: BAD POINTER.  Called by %s:%u\n", __FUNCTION__));
        goto error_handler;
    }

    /* Check the validity of message */
    if (strMsg == 0)
    {
        DbgPrint(("%s: BAD POINTER!\n", __FUNCTION__));
        goto error_handler;
    }

    if (ulSize != pmsg->ulSize)
    {
        DbgPrint(("%s: INVALID SIZE!\n", __FUNCTION__));
        goto error_handler;
    }

    /* Get Mutex */
    iRet = OS_SemTake(pmsg->mutex, OS_WAIT_FOREVER);
    if ( iRet != OS_OK ) 
    {
        DbgPrint(("%s: Failed to obtain lock\n",__FUNCTION__));
        goto error_handler;
    }



    /* If empty queue, wait */
    while (pmsg->ulHead == pmsg->ulTail)
    {
        iRet = OS_SemGive(pmsg->mutex);
        if ( iRet != OS_OK ) 
        {
            DbgPrint(("%s: Failed to release lock\n",__FUNCTION__));
            goto error_handler;
        }

        iRet = OS_SemTake( pmsg->waitOnBufferEmptySemaphore, iTimeout );
        if ( iRet == OS_TIMEOUT )
        {
            return ( iRet );
        }

        if ( iRet != OS_OK ) 
        {
            DbgPrint(("OS_MsgQSend: pthread_cond_wait ERROR\n"));
            goto error_handler;
        }

        waitedOnEmptySemaphore = TRUE;
        iRet = OS_SemTake(pmsg->mutex, OS_WAIT_FOREVER);
        if ( iRet != OS_OK ) 
        {
            DbgPrint(("%s: Failed to obtain lock\n",__FUNCTION__));
            goto error_handler;
        }


        if (( pmsg->ulHead == pmsg->ulTail ) && ( iTimeout != OS_WAIT_FOREVER ))
        {
            // If we waited for the message queue to not be empty, but somebody emptied it 
            // before we could read the message, and our timeout was not forever, give up with timeout
            // (otherwise would have to subtract time waited previously and only wait remaining timeout time)
            iRet = OS_TIMEOUT;
            break;
        }
    }

    /* if we didn't time out, receive it from the queue */
    if (OS_OK == iRet)
    {
        /* Receive it from the queue */
        if (iOption == OS_MSG_Q_TAIL)
        {
            /* Receive from tail of queue */
            pmsg->ulTail = ( (pmsg->ulTail + pmsg->ulDepth) % (pmsg->ulDepth + 1) );
            memcpy(strMsg, pmsg->pbBuffers + (pmsg->ulTail * pmsg->ulSize), pmsg->ulSize);
        }
        else
        {
            /* Receive from head of queue */
            memcpy(strMsg, pmsg->pbBuffers + (pmsg->ulHead * pmsg->ulSize), pmsg->ulSize);
            pmsg->ulHead = ( (pmsg->ulHead + 1) % (pmsg->ulDepth + 1) );
        }

        /* Signal any waiters */

        // buffer full semaphore count++
        iRet = OS_SemGive(pmsg->waitOnBufferFullSemaphore);
        if ( iRet != OS_OK ) 
        {
            DbgPrint(("%s: Broadcast Failed\n",__FUNCTION__));
        }


        // buffer empty semaphore count--
        if ( waitedOnEmptySemaphore == FALSE )
        {
            iRet = OS_SemTake(pmsg->waitOnBufferEmptySemaphore,OS_WAIT_FOREVER);
            if ( iRet != OS_OK ) 
            {
                DbgPrint(("%s: Broadcast Failed\n",__FUNCTION__));
            }
        }
    }

    /* Release mutex */
    iRet = OS_SemGive(pmsg->mutex);
    if (OS_OK != iRet)
    {
        DbgPrint(("%s: Failed to release lock\n",__FUNCTION__));
        goto error_handler;
    }

    /* Return success */
    return (iRet);

error_handler:

    /* Return failure */
    return (OS_FAILURE);
}

/**
 * Removes all messages from the queue.
 *
 * @param mqId - message queue identifier
 *
 * @retval OS_OK if successful, else an osapi error code.
 *
 * @remark
 *    None.
 *
 * @verified
 *    Yes.
 */
OS_STATUS OS_MsgQReset(OS_MSG_Q_ID mqId)
{
    OS_STATUS   status;
    OS_MBOX *pmsg = (OS_MBOX *)mqId;

    /* Check the validity of message queue */
    if (mqId == 0)
    {
        DbgPrint(("%s: BAD POINTER!\n", __FUNCTION__));
        goto error_handler;
    }

    /* Get Mutex */
    status = OS_SemTake(pmsg->mutex, OS_WAIT_FOREVER);
    if ( status != OS_OK ) 
    {
        DbgPrint(("%s: Failed to obtain lock\n",__FUNCTION__));
        goto error_handler;
    }

    /* reset queue */
    pmsg->ulHead = pmsg->ulTail;

    /* Signal any waiters */
    status = OS_SemGive(pmsg->waitOnBufferEmptySemaphore);
    if ( status != OS_OK ) 
    {
        DbgPrint(("%s: Broadcast Failed\n",__FUNCTION__));
    }
    status = OS_SemGive(pmsg->waitOnBufferFullSemaphore);
    if ( status != OS_OK ) 
    {
        DbgPrint(("%s: Broadcast Failed\n",__FUNCTION__));
    }

    /* Release mutex */
    status = OS_SemGive(pmsg->mutex);
    if ( status != OS_OK ) 
    {
        DbgPrint(("%s: Failed to release lock\n",__FUNCTION__));
        goto error_handler;
    }

    /* Return success */
    return (OS_OK);

error_handler:

    /* Return failure */
    return (OS_FAILURE);
}

/**
 * OS Message Queue Number of Messages function.
 *
 * @param
 *    OS_MSG_Q_ID mqId - message queue identifier
 *
 * @retval
 *    Returns the number of messages in the given message queue.
 *
 * @remark
 *    None.
 *
 * @verified
 *    Yes.
 */
int OS_MsgQNumMsgs(OS_MSG_Q_ID mqId)
{
    OS_MBOX *pmsg        = (OS_MBOX *)mqId;
    int     num_messages = 0;
    OS_STATUS           status;

    /* Check the validity of message queue */
    if (mqId == 0)
    {
        DbgPrint(("%s: BAD POINTER!\n", __FUNCTION__));
        return (OS_FAILURE);
    }

    /* Get Mutex */
    status = OS_SemTake(pmsg->mutex, OS_WAIT_FOREVER);
    if ( status == OS_OK ) 
    {
        if (pmsg->ulHead > pmsg->ulTail)
        {
            num_messages = pmsg->ulTail + pmsg->ulDepth + 1 - pmsg->ulHead;
        }
        else
        {
            num_messages = pmsg->ulTail - pmsg->ulHead;
        }

        /* Release mutex */
        status = OS_SemGive(pmsg->mutex);
        if ( status != OS_OK ) 
        {
            DbgPrint(("OS_SemGive failed\n"));
            return (OS_FAILURE);
        }
    }
    else
    {
        num_messages = OS_FAILURE;
    }

    return (num_messages);
}