pipedrv.c

VxWorks操作系统内核源代码

* EMFILE                    - pipe still has other openings
* EBUSY                     - pipe is selected by at least one pending task
*/

STATUS pipeDevDelete
    (
    char * name,	/* name of pipe to be deleted */
    BOOL   force	/* if TRUE, force pipe deletion */
    )
    {
    FAST PIPE_DEV *   pPipeDev;
    char *            pTail = NULL;
    SEL_WAKEUP_NODE * pNode = NULL;

    /* can't be called from ISR */

    if (INT_RESTRICT () != OK)
        {
        return (ERROR);
        }

    /* driver must be initialized */

    if (pipeDrvNum == ERROR)
        {
        errno = S_ioLib_NO_DRIVER;
        return (ERROR);
        }

    /* get pointer to pipe device descriptor */

    if ((pPipeDev = (PIPE_DEV *) iosDevFind (name, &pTail)) == NULL)
        {
	return (ERROR);
        }

    /* if not forced, check for other opens and non-empty select list */

    if (!force)
        {
        if (pPipeDev->numOpens != 0)
            {
            errno = EMFILE;
            return (ERROR);
            }

        if (selWakeupListLen (&pPipeDev->selWakeupList) != 0)
            {
            errno = EBUSY;
            return (ERROR);
            }
        }

    /* I/O device no longer in system */

    iosDevDelete (&pPipeDev->devHdr);

    /* force clearing of any select list */

    if(force && (selWakeupListLen (&pPipeDev->selWakeupList) != 0))
        {
        pNode = (SEL_WAKEUP_NODE *)lstFirst ((LIST *)&pPipeDev->selWakeupList);
        do
            {
            selNodeDelete (&pPipeDev->selWakeupList, pNode);
            } while ((pNode = (SEL_WAKEUP_NODE *) lstNext ((NODE *)pNode))
                           != NULL);
        lstFree ((LIST *)&pPipeDev->selWakeupList);
        selWakeupListTerm (&pPipeDev->selWakeupList);
        }

    /* terminate message queue */

    msgQTerminate (&pPipeDev->msgQ);

    /* free pipe memory */

    KHEAP_FREE ((char *)pPipeDev);

    return (OK);
    }

/* routines supplied to I/O system */

/*******************************************************************************
*
* pipeOpen - open a pipe file
*
* This routine is called to open a pipe.  It returns a pointer to the
* device.  This routine is normally reached only via the I/O system.
*
* RETURNS  pPipeDev or ERROR if pipe has not been created by pipeDevCreate().
*/

LOCAL int pipeOpen
    (
    PIPE_DEV * pPipeDev,	/* pipe descriptor */
    char *     name,
    int        flags,
    int        mode
    )
    {
    if ((name != NULL) && (strlen (name) > 0))
	{
	/* Only the first part of the name match with the driver's name */ 

	errnoSet (S_ioLib_NO_DEVICE_NAME_IN_PATH);
	return (ERROR);
	}
    else
	{
	/* track number of openings to pipe */

	++pPipeDev->numOpens;
    	return ((int) pPipeDev);
	}
    }

/*******************************************************************************
*
* pipeClose - close a pipe file
*
* This routine is called to close a pipe.  This routine is normally reached
* only via the I/O system.
*
* RETURNS  OK or ERROR if NULL pipe device pointer.
*/

LOCAL int pipeClose
    (
    PIPE_DEV * pPipeDev		/* pipe descriptor */
    )
    {
    if (pPipeDev != NULL)
	{
	/* decrement the open counter, but not past zero */ 

	if (pPipeDev->numOpens > 0)
	  --pPipeDev->numOpens;
	return (OK);
	}
    else
	{
    	return (ERROR);
	}
    }

/*******************************************************************************
*
* pipeRead - read bytes from a pipe
*
* This routine reads up to maxbytes bytes of the next message in the pipe.
* If the message is too long, the additional bytes are just discarded.
*
* RETURNS:
*  number of bytes actually read;
*  will be between 1 and maxbytes, or ERROR
*/

LOCAL int pipeRead
    (
    FAST PIPE_DEV * pPipeDev,	/* pointer to pipe descriptor */
    char *          buffer,	/* buffer to receive bytes */
    unsigned int    maxbytes	/* max number of bytes to copy into buffer */
    )
    {
    int nbytes;

    /* wait for something to be in pipe */

    nbytes = msgQReceive (&pPipeDev->msgQ, buffer, maxbytes, WAIT_FOREVER);

    if (nbytes == ERROR)
	return (ERROR);

    /* wake up any select-blocked writers */

    selWakeupAll (&pPipeDev->selWakeupList, SELWRITE);

    return (nbytes);
    }

/*******************************************************************************
*
* pipeWrite - write bytes to a pipe
*
* This routine writes a message of `nbytes' to the pipe.
*
* RETURNS: number of bytes written or ERROR
*/

LOCAL int pipeWrite
    (
    FAST PIPE_DEV * pPipeDev,	/* pointer to pipe descriptor */
    char *          buffer,	/* buffer from which to copy bytes */
    int             nbytes	/* number of bytes to copy from buffer */
    )
    {
    if (!INT_CONTEXT ())
	TASK_LOCK ();				/* LOCK PREEMPTION */

    /* 
     * We lock preemption so after we send the message we can guarantee that
     * we get to the selWakeupAll() call before unblocking any readers.  This
     * is to avoid a race in which a higher priority reader of the pipe is
     * unblocked by the msgQSend() below and subsequently enters and blocks
     * in a call to select(), only to be inadvertently awakened when we return
     * here and call selWakeupAll().  To minimize preemption latency we
     * release the preemption lock after we obtain the selWakeupList mutual
     * exclusion semaphore.  This semaphore is a mutual exclusion semaphore
     * which allows recursive takes.  Avoiding a preemption lock by utilizing
     * the selWakeupList semaphore as the only means of mutual exclusion is
     * not viable because deadlock can occur by virtue of the fact that
     * msgQSend() can block if the the message queue is full at which time a
     * call to select() could block waiting for the listMutex instead of
     * returning that a read is OK.  A problem this approach does not account
     * for is the possibility that the selWakeupList semaphore is unavailable
     * when the semTake() is attempted below.  If this were the case, the
     * task could be preempted and therefore be vulnerable to the same
     * scenario outlined above.  
     */

    if (msgQSend (&pPipeDev->msgQ, buffer, (UINT) nbytes,
		  INT_CONTEXT() ? NO_WAIT : WAIT_FOREVER, MSG_PRI_NORMAL) != OK)
	{
	if (!INT_CONTEXT ())
	    TASK_UNLOCK ();			/* UNLOCK PREEMPTION */
	return (ERROR);
	}

    if (!INT_CONTEXT ())
	{
	semTake (&pPipeDev->selWakeupList.listMutex, WAIT_FOREVER);
	TASK_UNLOCK ();				/* UNLOCK PREEMPTION */
	}

    /* wake up any select-blocked readers */

    selWakeupAll (&pPipeDev->selWakeupList, SELREAD);

    if (!INT_CONTEXT ())
	semGive (&pPipeDev->selWakeupList.listMutex);

    return (nbytes);
    }

/*******************************************************************************
*
* pipeIoctl - do device specific control function
*
* The ioctl requests recognized are FIONREAD, FIONMSGS, and FIOFLUSH.
*
* RETURNS:
*  OK and `argptr' gets number of bytes in pipe, or
*  ERROR if request is not FIONREAD, FIONMSGS, or FIOFLUSH.
*/

LOCAL STATUS pipeIoctl
    (
    FAST PIPE_DEV *pPipeDev,	/* pointer to pipe descriptor */
    int           request,	/* ioctl code */
    int           *argptr	/* where to send answer */
    )
    {
    STATUS	status = OK;
    MSG_Q_INFO	msgQInfo;
    SEL_WAKEUP_NODE * wakeNode = (SEL_WAKEUP_NODE *) argptr;

    switch (request)
	{
	case FIONREAD:
	    /* number of bytes in 1st message in the queue */

	    bzero ((char *) &msgQInfo, sizeof (msgQInfo));
	    msgQInfo.msgListMax	= 1;
	    msgQInfo.msgLenList	= argptr;
	    *argptr = 0;
	    msgQInfoGet (&pPipeDev->msgQ, &msgQInfo);
	    break;

	case FIONMSGS:
	    /* number of messages in pipe */

	    *argptr = msgQNumMsgs (&pPipeDev->msgQ);
	    break;

	case FIOFLUSH:
	    /* discard all outstanding messages */

	    taskLock ();
	    while (msgQReceive (&pPipeDev->msgQ, (char *) NULL, 0, NO_WAIT) !=
									ERROR)
		;
	    taskUnlock ();
	    break;

	case FIOSELECT:
	    selNodeAdd (&pPipeDev->selWakeupList, (SEL_WAKEUP_NODE *) argptr);

	    switch (wakeNode->type)
		{
		case SELREAD:
		    if (msgQNumMsgs (&pPipeDev->msgQ) > 0)
		        selWakeup ((SEL_WAKEUP_NODE *) argptr);
		    break;
		    
		case SELWRITE:
		    if (pPipeDev->msgQ.maxMsgs >
			msgQNumMsgs (&pPipeDev->msgQ))
		        selWakeup ((SEL_WAKEUP_NODE *) argptr);
                    break;
		}
            break;

        case FIOUNSELECT:
	    selNodeDelete (&pPipeDev->selWakeupList, (SEL_WAKEUP_NODE *)argptr);
            break;

	default:
	    status = ERROR;
	    errnoSet (S_ioLib_UNKNOWN_REQUEST);
	    break;
	}

    return (status);
    }