w16sock.c

Netscape NSPR库源码

                pri = pq->thr->priority;
                pq->thr->state = _PR_RUNNABLE;

                _PR_RUNQ_LOCK(cpu);
                _PR_ADD_RUNQ(pq->thr, cpu, pri);
                _PR_RUNQ_UNLOCK(cpu);
            }
            _PR_THREAD_UNLOCK(pq->thr);
        } else {
            if (pq->timeout < _PR_IOQ_TIMEOUT(me->cpu))
                _PR_IOQ_TIMEOUT(me->cpu) = pq->timeout;
            if (_PR_IOQ_MAX_OSFD(me->cpu) < pq_max_osfd)
                _PR_IOQ_MAX_OSFD(me->cpu) = pq_max_osfd;
        }
    }
} /* end FindBadFDs() */

/*
** Called by the scheduler when there is nothing to do. This means that
** all threads are blocked on some monitor somewhere.
**
** Pause the current CPU. longjmp to the cpu's pause stack
*/
PRInt32 _PR_MD_PAUSE_CPU( PRIntervalTime ticks)
{
    PRThread *me = _MD_CURRENT_THREAD();
    struct timeval timeout, *tvp;
    fd_set r, w, e;
    fd_set *rp, *wp, *ep;
    PRInt32 max_osfd, nfd;
    PRInt32 rv;
    PRCList *q;
    PRUint32 min_timeout;

    PR_ASSERT(_PR_MD_GET_INTSOFF() != 0);

    /*
     * assigment of fd_sets
     */
    r = _PR_FD_READ_SET(me->cpu);
    w = _PR_FD_WRITE_SET(me->cpu);
    e = _PR_FD_EXCEPTION_SET(me->cpu);

    rp = &r;
    wp = &w;
    ep = &e;

    max_osfd = _PR_IOQ_MAX_OSFD(me->cpu) + 1;
    min_timeout = _PR_IOQ_TIMEOUT(me->cpu);
    /*
    ** Compute the minimum timeout value: make it the smaller of the
    ** timeouts specified by the i/o pollers or the timeout of the first
    ** sleeping thread.
    */
    q = _PR_SLEEPQ(me->cpu).next;

    if (q != &_PR_SLEEPQ(me->cpu)) {
        PRThread *t = _PR_THREAD_PTR(q);

        if (t->sleep < min_timeout) {
            min_timeout = t->sleep;
        }
    }
    if (min_timeout > ticks) {
        min_timeout = ticks;
    }

    if (min_timeout == PR_INTERVAL_NO_TIMEOUT) {
        tvp = NULL;
    } else {
        timeout.tv_sec = PR_IntervalToSeconds(min_timeout);
        timeout.tv_usec = PR_IntervalToMicroseconds(min_timeout)
            % PR_USEC_PER_SEC;
        tvp = &timeout;
    }

    _PR_MD_IOQ_UNLOCK();
    _MD_CHECK_FOR_EXIT();
    /*
     * check for i/o operations
     */

    nfd = _MD_SELECT(max_osfd, rp, wp, ep, tvp);

    _MD_CHECK_FOR_EXIT();
    _PR_MD_IOQ_LOCK();
    /*
    ** Notify monitors that are associated with the selected descriptors.
    */
    if (nfd > 0) {
        q = _PR_IOQ(me->cpu).next;
        _PR_IOQ_MAX_OSFD(me->cpu) = -1;
        _PR_IOQ_TIMEOUT(me->cpu) = PR_INTERVAL_NO_TIMEOUT;
        while (q != &_PR_IOQ(me->cpu)) {
            PRPollQueue *pq = _PR_POLLQUEUE_PTR(q);
            PRBool notify = PR_FALSE;
            _PRWin16PollDesc *pds = pq->pds;
            _PRWin16PollDesc *epds = pds + pq->npds;
            PRInt32 pq_max_osfd = -1;

            q = q->next;
            for (; pds < epds; pds++) {
                PRInt32 osfd = pds->osfd;
                PRInt16 in_flags = pds->in_flags;
                PRInt16 out_flags = 0;
                PR_ASSERT(osfd >= 0 || in_flags == 0);
                if ((in_flags & PR_POLL_READ) && FD_ISSET(osfd, rp)) {
                    out_flags |= PR_POLL_READ;
                }
                if ((in_flags & PR_POLL_WRITE) && FD_ISSET(osfd, wp)) {
                    out_flags |= PR_POLL_WRITE;
                }
                if ((in_flags & PR_POLL_EXCEPT) && FD_ISSET(osfd, ep)) {
                    out_flags |= PR_POLL_EXCEPT;
                }
                pds->out_flags = out_flags;
                if (out_flags) {
                    notify = PR_TRUE;
                }
                if (osfd > pq_max_osfd) {
                    pq_max_osfd = osfd;
                }
            }
            if (notify == PR_TRUE) {
                PRIntn pri;
                PRThread *thred;

                PR_REMOVE_LINK(&pq->links);
                pq->on_ioq = PR_FALSE;

                /*
                 * Decrement the count of descriptors for each desciptor/event
                 * because this I/O request is being removed from the
                 * ioq
                 */
                pds = pq->pds;
                for (; pds < epds; pds++) {
                    PRInt32 osfd = pds->osfd;
                    PRInt16 in_flags = pds->in_flags;
                    PR_ASSERT(osfd >= 0 || in_flags == 0);
                    if (in_flags & PR_POLL_READ) {
                        if (--(_PR_FD_READ_CNT(me->cpu))[osfd] == 0)
                            FD_CLR(osfd, &_PR_FD_READ_SET(me->cpu));
                    }
                    if (in_flags & PR_POLL_WRITE) {
                        if (--(_PR_FD_WRITE_CNT(me->cpu))[osfd] == 0)
                            FD_CLR(osfd, &_PR_FD_WRITE_SET(me->cpu));
                    }
                    if (in_flags & PR_POLL_EXCEPT) {
                        if (--(_PR_FD_EXCEPTION_CNT(me->cpu))[osfd] == 0)
                            FD_CLR(osfd, &_PR_FD_EXCEPTION_SET(me->cpu));
                    }
                }
                 thred = pq->thr;
                _PR_THREAD_LOCK(thred);
                if (pq->thr->flags & (_PR_ON_PAUSEQ|_PR_ON_SLEEPQ)) {
                    _PRCPU *cpu = thred->cpu;
                    _PR_SLEEPQ_LOCK(pq->thr->cpu);
                    _PR_DEL_SLEEPQ(pq->thr, PR_TRUE);
                    _PR_SLEEPQ_UNLOCK(pq->thr->cpu);

                    pri = pq->thr->priority;
                    pq->thr->state = _PR_RUNNABLE;

                    pq->thr->cpu = cpu;
                    _PR_RUNQ_LOCK(cpu);
                    _PR_ADD_RUNQ(pq->thr, cpu, pri);
                    _PR_RUNQ_UNLOCK(cpu);
                    if (_pr_md_idle_cpus > 1)
                        _PR_MD_WAKEUP_WAITER(thred);
                }
                _PR_THREAD_UNLOCK(thred);
            } else {
                if (pq->timeout < _PR_IOQ_TIMEOUT(me->cpu))
                    _PR_IOQ_TIMEOUT(me->cpu) = pq->timeout;
                if (_PR_IOQ_MAX_OSFD(me->cpu) < pq_max_osfd)
                    _PR_IOQ_MAX_OSFD(me->cpu) = pq_max_osfd;
            }
        }
    } else if (nfd < 0) {
        if ( WSAGetLastError() == WSAENOTSOCK )
        {
            FindBadFDs();
        } else {
            PR_LOG(_pr_io_lm, PR_LOG_MAX, ("select() failed with errno %d",
                errno));
        }
    }
    _PR_MD_IOQ_UNLOCK();
    return(0);
    
} /* end _PR_MD_PAUSE_CPU() */


/*
** _MD_pr_poll() -- Implement MD polling
**
** The function was snatched (re-used) from the unix implementation.
** 
** The native thread stuff was deleted.
** The pollqueue is instantiated on the mdthread structure
** to keep the stack frame from being corrupted when this
** thread is waiting on the poll.
**
*/
extern PRInt32 
_MD_PR_POLL(PRPollDesc *pds, PRIntn npds,
                        PRIntervalTime timeout)
{
    PRPollDesc *pd, *epd;
    PRInt32 n, err, pdcnt;
    PRIntn is;
    _PRWin16PollDesc *spds, *spd;
    PRThread *me = _PR_MD_CURRENT_THREAD();
    PRPollQueue *pq;

    pq = &me->md.thr_pq;
    
    /*
     * XXX
     *   PRPollDesc has a PRFileDesc field, fd, while the IOQ
     *   is a list of PRPollQueue structures, each of which contains
     *   a _PRWin16PollDesc. A _PRWin16PollDesc struct contains
     *   the OS file descriptor, osfd, and not a PRFileDesc.
     *   So, we have allocate memory for _PRWin16PollDesc structures,
     *   copy the flags information from the pds list and have pq
     *   point to this list of _PRWin16PollDesc structures.
     *
     *   It would be better if the memory allocation can be avoided.
     */

    spds = (_PRWin16PollDesc*) PR_MALLOC(npds * sizeof(_PRWin16PollDesc));
    if (!spds) {
        PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
           return -1;
    }
    spd = spds;

    _PR_INTSOFF(is);
    _PR_MD_IOQ_LOCK();
       _PR_THREAD_LOCK(me);

    if (_PR_PENDING_INTERRUPT(me)) {
        me->flags &= ~_PR_INTERRUPT;
        PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
        _PR_THREAD_UNLOCK(me);
        _PR_MD_IOQ_UNLOCK();
        PR_DELETE(spds);
           return -1;
    }

    pdcnt = 0;
    for (pd = pds, epd = pd + npds; pd < epd; pd++) {
        PRInt32 osfd;
        PRInt16 in_flags = pd->in_flags;
        PRFileDesc *bottom = pd->fd;

        if ((NULL == bottom) || (in_flags == 0)) {
            continue;
        }
        while (bottom->lower != NULL) {
            bottom = bottom->lower;
        }
        osfd = bottom->secret->md.osfd;

        PR_ASSERT(osfd >= 0 || in_flags == 0);

        spd->osfd = osfd;
        spd->in_flags = pd->in_flags;
        spd++;
        pdcnt++;

        if (in_flags & PR_POLL_READ)  {
            FD_SET(osfd, &_PR_FD_READ_SET(me->cpu));
            _PR_FD_READ_CNT(me->cpu)[osfd]++;
        }
        if (in_flags & PR_POLL_WRITE) {
            FD_SET(osfd, &_PR_FD_WRITE_SET(me->cpu));
            (_PR_FD_WRITE_CNT(me->cpu))[osfd]++;
        }
        if (in_flags & PR_POLL_EXCEPT) {
            FD_SET(osfd, &_PR_FD_EXCEPTION_SET(me->cpu));
            (_PR_FD_EXCEPTION_CNT(me->cpu))[osfd]++;
        }
        if (osfd > _PR_IOQ_MAX_OSFD(me->cpu))
            _PR_IOQ_MAX_OSFD(me->cpu) = osfd;
    }
    if (timeout < _PR_IOQ_TIMEOUT(me->cpu))
        _PR_IOQ_TIMEOUT(me->cpu) = timeout;


    pq->pds = spds;
    pq->npds = pdcnt;

    pq->thr = me;
    pq->on_ioq = PR_TRUE;
    pq->timeout = timeout;
    _PR_ADD_TO_IOQ((*pq), me->cpu);
    _PR_SLEEPQ_LOCK(me->cpu);
    _PR_ADD_SLEEPQ(me, timeout);
    me->state = _PR_IO_WAIT;
    me->io_pending = PR_TRUE;
    me->io_suspended = PR_FALSE;
    _PR_SLEEPQ_UNLOCK(me->cpu);
    _PR_THREAD_UNLOCK(me);
    _PR_MD_IOQ_UNLOCK();

    _PR_MD_WAIT(me, timeout);

    me->io_pending = PR_FALSE;
    me->io_suspended = PR_FALSE;

    /*
     * Copy the out_flags from the _PRWin16PollDesc structures to the
     * user's PRPollDesc structures and free the allocated memory
     */
    spd = spds;
    for (pd = pds, epd = pd + npds; pd < epd; pd++) {
        if ((NULL == pd->fd) || (pd->in_flags == 0)) {
            pd->out_flags = 0;
            continue;
        }
        pd->out_flags = spd->out_flags;
        spd++;
    }
    PR_DELETE(spds);

    /*
    ** If we timed out the pollq might still be on the ioq. Remove it
    ** before continuing.
    */
    if (pq->on_ioq) {
        _PR_INTSOFF(is);
        _PR_MD_IOQ_LOCK();
        /*
         * Need to check pq.on_ioq again
         */
        if (pq->on_ioq == PR_TRUE) {
            PR_REMOVE_LINK(&pq->links);
            for (pd = pds, epd = pd + npds; pd < epd; pd++) {
                PRInt32 osfd;
                PRInt16 in_flags = pd->in_flags;
                PRFileDesc *bottom = pd->fd;

                if ((NULL == bottom) || (in_flags == 0)) {
                    continue;
                }
                while (bottom->lower != NULL) {
                    bottom = bottom->lower;
                }
                osfd = bottom->secret->md.osfd;
                PR_ASSERT(osfd >= 0 || in_flags == 0);
                if (in_flags & PR_POLL_READ)  {
                    if (--(_PR_FD_READ_CNT(me->cpu))[osfd] == 0)
                        FD_CLR(osfd, &_PR_FD_READ_SET(me->cpu));
                }
                if (in_flags & PR_POLL_WRITE) {
                    if (--(_PR_FD_WRITE_CNT(me->cpu))[osfd] == 0)
                            FD_CLR(osfd, &_PR_FD_WRITE_SET(me->cpu));
                }
                if (in_flags & PR_POLL_EXCEPT) {
                    if (--(_PR_FD_EXCEPTION_CNT(me->cpu))[osfd] == 0)
                            FD_CLR(osfd, &_PR_FD_EXCEPTION_SET(me->cpu));
                }
            }
        }
        _PR_MD_IOQ_UNLOCK();
        _PR_INTSON(is);
    }
    if (_PR_PENDING_INTERRUPT(me)) {
        me->flags &= ~_PR_INTERRUPT;
        PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
           return -1;
    } else {
        n = 0;
        if (pq->on_ioq == PR_FALSE) {
            /* Count the number of ready descriptors */
            while (--npds >= 0) {
            if (pds->out_flags) {
                n++;
            }
                pds++;
            }
        }
        return n;
    }
} /* end _MD_pr_poll() */