msgqlib.c

vxworks源码源码解读是学习vxworks的最佳途径

    return (msgQDestroy (msgQId, TRUE));
    }

/*******************************************************************************
*
* msgQDestroy - destroy message queue
*
* RETURNS: OK or ERROR.
*
* ERRNO: S_distLib_NO_OBJECT_DESTROY
*
*/

LOCAL STATUS msgQDestroy
    (
    MSG_Q_ID msgQId,    /* message queue to destroy */
    BOOL     dealloc    /* deallocate memory associated with message queue */
    )
    {
    Q_JOB_NODE *pNode;
    FAST int	timeout;
    FAST int	nMsgs;

    if (ID_IS_SHARED (msgQId))  		/* message Q is shared?*/
	{
		if (ID_IS_DISTRIBUTED (msgQId)) /* message queue is distributed? */
			{
			errno = S_distLib_NO_OBJECT_DESTROY;
		 	return (ERROR);             /* cannot delete distributed msgQ */
			}
	
		errno = S_smObjLib_NO_OBJECT_DESTROY;
        return (ERROR);				/* cannot delete sm. msgQ */
	}

    if (INT_RESTRICT () != OK)			/* restrict isr use */
	return (ERROR);

    TASK_SAFE ();				/* TASK SAFE */
    TASK_LOCK ();				/* LOCK PREEMPTION */

#ifdef WV_INSTRUMENTATION

    /* Indicate that msgQDelete has been initiated */

    /* windview - level 1 event logging routine */
    EVT_OBJ_1 (OBJ, msgQId, msgQClassId, EVENT_MSGQDELETE, msgQId);
#endif

    if (OBJ_VERIFY (msgQId, msgQClassId) != OK)	/* validate message queue id */
	{
	TASK_UNLOCK ();				/* UNLOCK PREEMPTION */
	TASK_UNSAFE ();				/* TASK UNSAFE */
	return (ERROR);
	}

    objCoreTerminate (&msgQId->objCore);	/* INVALIDATE */

#ifdef WV_INSTRUMENTATION

    /*  Indicate that the msgQDelete has succeeded (before TASK_UNLOCK, as
     *  that causes unnecessary WV parser confusion).
     */

    /* windview - level 2 instrumentation
     * EVENT_OBJ_MSGDELETE needs to return the msgQId so MSG_OFFSET is
     * used to calulate the msgQId from the pQHead
     */
    EVT_TASK_1 (EVENT_OBJ_MSGDELETE, msgQId);
#endif

    TASK_UNLOCK ();				/* UNLOCK PREEMPTION */


    /* gobble up all messages in the message and free queues */

    timeout = NO_WAIT;		/* first time through gobble without waiting */
    nMsgs = 0;

    while (nMsgs < msgQId->maxMsgs)
	{
	while (((pNode = qJobGet (&msgQId->freeQ, timeout)) != NULL) &&
	       (pNode != (Q_JOB_NODE *) NONE))
	    nMsgs++;

	while (((pNode = qJobGet (&msgQId->msgQ, timeout)) != NULL) &&
	       (pNode != (Q_JOB_NODE *) NONE))
	    nMsgs++;

	timeout = 1;		/* after first time, wait a bit */
	}

    /* terminate both the queues */

    qJobTerminate (&msgQId->msgQ);
    qJobTerminate (&msgQId->freeQ);

    if (dealloc)
	objFree (msgQClassId, (char *) msgQId);

    TASK_UNSAFE ();				/* TASK UNSAFE */

    return (OK);
    }

/*******************************************************************************
*
* msgQSend - send a message to a message queue
*
* This routine sends the message in <buffer> of length <nBytes> to the message
* queue <msgQId>.  If any tasks are already waiting to receive messages
* on the queue, the message will immediately be delivered to the first
* waiting task.  If no task is waiting to receive messages, the message
* is saved in the message queue.
*
* The <timeout> parameter specifies the number of ticks to wait for free
* space if the message queue is full.  The <timeout> parameter can also have 
* the following special values:
* .iP "NO_WAIT  (0)" 8
* return immediately, even if the message has not been sent.  
* .iP "WAIT_FOREVER  (-1)"
* never time out.
* .LP
*
* The <priority> parameter specifies the priority of the message being sent.
* The possible values are:
* .iP "MSG_PRI_NORMAL  (0)" 8
* normal priority; add the message to the tail of the list of queued 
* messages.
* .iP "MSG_PRI_URGENT  (1)"
* urgent priority; add the message to the head of the list of queued messages.
* .LP
*
* USE BY INTERRUPT SERVICE ROUTINES
* This routine can be called by interrupt service routines as well as
* by tasks.  This is one of the primary means of communication
* between an interrupt service routine and a task.  When called from an
* interrupt service routine, <timeout> must be NO_WAIT.
*
* RETURNS: OK or ERROR.
*
* ERRNO:  S_distLib_NOT_INITIALIZED, S_objLib_OBJ_ID_ERROR,
*         S_objLib_OBJ_DELETED, S_objLib_OBJ_UNAVAILABLE,
*         S_objLib_OBJ_TIMEOUT, S_msgQLib_INVALID_MSG_LENGTH,
*         S_msgQLib_NON_ZERO_TIMEOUT_AT_INT_LEVEL
*
* SEE ALSO: msgQSmLib
*/

STATUS msgQSend
    (
    FAST MSG_Q_ID       msgQId,         /* message queue on which to send */
    char *              buffer,         /* message to send */
    FAST UINT           nBytes,         /* length of message */
    int                 timeout,        /* ticks to wait */
    int                 priority        /* MSG_PRI_NORMAL or MSG_PRI_URGENT */
    )
    {
    FAST MSG_NODE *	pMsg;

    if (ID_IS_SHARED (msgQId))			/* message Q is shared? */
        {
		if (ID_IS_DISTRIBUTED (msgQId)) /* message queue is distributed? */
			{
			if (msgQDistSendRtn == NULL)
				{
				errno = S_distLib_NOT_INITIALIZED; 
				return (ERROR);
				}
		 	return ((*msgQDistSendRtn) (msgQId, buffer, nBytes,
			 	timeout, WAIT_FOREVER, priority));
			}

        if (msgQSmSendRtn == NULL)
            {
            errno = S_smObjLib_NOT_INITIALIZED;
            return (ERROR);
            }

        return ((*msgQSmSendRtn) (SM_OBJ_ID_TO_ADRS (msgQId), buffer, nBytes,
				  timeout, priority));
	}

    /* message queue is local */

    if (!INT_CONTEXT ())
	TASK_LOCK ();
    else
	{
	if (timeout != 0)
	    {
	    errnoSet (S_msgQLib_NON_ZERO_TIMEOUT_AT_INT_LEVEL);
	    return (ERROR);
	    }
	}

restart:
    if (OBJ_VERIFY (msgQId, msgQClassId) != OK)
	{
	if (!INT_CONTEXT ())
	    TASK_UNLOCK ();
	return (ERROR);
	}

#ifdef WV_INSTRUMENTATION
    /* windview - level 1 event logging routine */
    EVT_OBJ_5 (OBJ, msgQId, msgQClassId, EVENT_MSGQSEND, msgQId, 
	       buffer, nBytes, timeout, priority);
#endif

    if (nBytes > msgQId->maxMsgLength)
	{
	if (!INT_CONTEXT ())
	    TASK_UNLOCK ();
	errnoSet (S_msgQLib_INVALID_MSG_LENGTH);
	return (ERROR);
	}

    pMsg = (MSG_NODE *) qJobGet (&msgQId->freeQ, timeout);

    if (pMsg == (MSG_NODE *) NONE)
	{
	timeout = SIG_TIMEOUT_RECALC(timeout);
	goto restart;
	}

    if (pMsg == NULL)
	{
	msgQId->sendTimeouts++;
	if (!INT_CONTEXT ())
	    TASK_UNLOCK ();
	return (ERROR);
	}

    pMsg->msgLength = nBytes;
    bcopy (buffer, MSG_NODE_DATA (pMsg), (int) nBytes);

    qJobPut (&msgQId->msgQ, &pMsg->node, priority);

    if (!INT_CONTEXT ())
	TASK_UNLOCK ();

    return (OK);
    }

/*******************************************************************************
*
* msgQReceive - receive a message from a message queue
*
* This routine receives a message from the message queue <msgQId>.
* The received message is copied into the specified <buffer>, which is
* <maxNBytes> in length.  If the message is longer than <maxNBytes>,
* the remainder of the message is discarded (no error indication
* is returned).
*
* The <timeout> parameter specifies the number of ticks to wait for 
* a message to be sent to the queue, if no message is available when
* msgQReceive() is called.  The <timeout> parameter can also have 
* the following special values: 
* .iP "NO_WAIT  (0)" 8
* return immediately, even if the message has not been sent.  
* .iP "WAIT_FOREVER  (-1)"
* never time out.
* .LP
*
* WARNING: This routine must not be called by interrupt service routines.
*
* RETURNS:
* The number of bytes copied to <buffer>, or ERROR.
*
* ERRNO: S_distLib_NOT_INITIALIZED, S_smObjLib_NOT_INITIALIZED,
*        S_objLib_OBJ_ID_ERROR, S_objLib_OBJ_DELETED,
*        S_objLib_OBJ_UNAVAILABLE, S_objLib_OBJ_TIMEOUT,
*        S_msgQLib_INVALID_MSG_LENGTH
*
* SEE ALSO: msgQSmLib
*/

int msgQReceive
    (
    FAST MSG_Q_ID       msgQId,         /* message queue from which to receive */
    char *              buffer,         /* buffer to receive message */
    UINT                maxNBytes,      /* length of buffer */
    int                 timeout         /* ticks to wait */
    )
    {
    FAST MSG_NODE *	pMsg;
    FAST int		bytesReturned;

    if (ID_IS_SHARED (msgQId))			/* message Q is shared? */
        {
		if (ID_IS_DISTRIBUTED (msgQId)) /* message queue is distributed? */
			{
			if (msgQDistReceiveRtn == NULL)
				{
				errno = S_distLib_NOT_INITIALIZED;
			 	return (ERROR);
				}
			return ((*msgQDistReceiveRtn) (msgQId, buffer,
				maxNBytes, timeout, WAIT_FOREVER));
			}

        if (msgQSmReceiveRtn == NULL)
            {
            errno = S_smObjLib_NOT_INITIALIZED;
            return (ERROR);
            }

        return ((*msgQSmReceiveRtn) (SM_OBJ_ID_TO_ADRS (msgQId), buffer,
				     maxNBytes, timeout));
	}

    /* message queue is local */

    /* even though maxNBytes is unsigned, check for < 0 to catch possible
     * caller errors
     */
    if ((int) maxNBytes < 0)
	{
	errnoSet (S_msgQLib_INVALID_MSG_LENGTH);
	return (ERROR);
	}

    TASK_LOCK ();

restart:
    if (OBJ_VERIFY (msgQId, msgQClassId) != OK)
	{
	TASK_UNLOCK ();
	return (ERROR);
	}

#ifdef WV_INSTRUMENTATION
    /* windview - level 1 event logging routine */
    EVT_OBJ_4 (OBJ, msgQId, msgQClassId, EVENT_MSGQRECEIVE, msgQId, 
	       buffer, maxNBytes, timeout);
#endif

    pMsg = (MSG_NODE *) qJobGet (&msgQId->msgQ, timeout);

    if (pMsg == (MSG_NODE *) NONE)
	{
	timeout = SIG_TIMEOUT_RECALC(timeout);
	goto restart;
	}

    if (pMsg == NULL)
	{
	msgQId->recvTimeouts++;
	TASK_UNLOCK ();
	return (ERROR);
	}

    bytesReturned = min (pMsg->msgLength, maxNBytes);
    bcopy (MSG_NODE_DATA (pMsg), buffer, bytesReturned);

    qJobPut (&msgQId->freeQ, &pMsg->node, Q_JOB_PRI_DONT_CARE);

    TASK_UNLOCK ();

    return (bytesReturned);
    }

/*******************************************************************************
*
* msgQNumMsgs - get the number of messages queued to a message queue
*
* This routine returns the number of messages currently queued to a specified
* message queue.
*
* RETURNS:
* The number of messages queued, or ERROR.
*
* ERRNO:  S_distLib_NOT_INITIALIZED, S_smObjLib_NOT_INITIALIZED,
*         S_objLib_OBJ_ID_ERROR
*
* SEE ALSO: msgQSmLib
*/

int msgQNumMsgs
    (
    FAST MSG_Q_ID       msgQId          /* message queue to examine */
    )
    {
    if (ID_IS_SHARED (msgQId))			/* message Q is shared? */
        {
		if (ID_IS_DISTRIBUTED (msgQId)) /* message queue is distributed? */
			{
			if (msgQDistNumMsgsRtn == NULL)
				{
				errno = S_distLib_NOT_INITIALIZED;
				return (ERROR);
				}
			return ((*msgQDistNumMsgsRtn) (msgQId, WAIT_FOREVER));
			}

        if (msgQSmNumMsgsRtn == NULL)
            {
            errno = S_smObjLib_NOT_INITIALIZED;
            return (ERROR);
            }

        return ((*msgQSmNumMsgsRtn) (SM_OBJ_ID_TO_ADRS (msgQId)));
	}

    /* message queue is local */

    if (OBJ_VERIFY (msgQId, msgQClassId) != OK)
	return (ERROR);

    return (msgQId->msgQ.count);
    }