ntio.c

Netscape NSPR库源码

						 * must be running on the same cpu on which the
						 * I/O operation was issued.
						 */
						PR_ASSERT(!completed_io->md.thr_bound_cpu ||
					(completed_io->cpu == completed_io->md.thr_bound_cpu));

							if (!completed_io->md.thr_bound_cpu)
                            	completed_io->cpu = lockedCPU;
                            completed_io->state = _PR_RUNNABLE;
                            _PR_RUNQ_LOCK(completed_io->cpu);
                            _PR_ADD_RUNQ(completed_io, completed_io->cpu, pri);
                            _PR_RUNQ_UNLOCK(completed_io->cpu);
                        } else {
                            _PR_THREAD_UNLOCK(completed_io);
                        }
                    } else {
                        _PR_THREAD_UNLOCK(completed_io);
                    }
                }
            } else {
                /* For native threads, they are only notified through this loop
                 * when completing IO.  So, don't worry about this being a CVAR
                 * notification, because that is not possible.
                 */
                _PR_THREAD_LOCK(completed_io);
                completed_io->io_pending = PR_FALSE;
                if (completed_io->io_suspended == PR_FALSE) {
                    completed_io->state = _PR_RUNNABLE;
                    _PR_THREAD_UNLOCK(completed_io);
                    rv = ReleaseSemaphore(completed_io->md.blocked_sema,
                            1, NULL);
                    PR_ASSERT(0 != rv);
                } else {
                    _PR_THREAD_UNLOCK(completed_io);
                }
            }
        }

        awoken++;
        timeout = 0;   /* Don't block on subsequent trips through the loop */
    }

    /* never reached */
    return 0;
}

static PRStatus
_native_thread_md_wait(PRThread *thread, PRIntervalTime ticks)
{
    DWORD rv;
	PRUint32 msecs = (ticks == PR_INTERVAL_NO_TIMEOUT) ?
		INFINITE : PR_IntervalToMilliseconds(ticks);

	/*
	 * thread waiting for a cvar or a joining thread
	 */
	rv = WaitForSingleObject(thread->md.blocked_sema, msecs);
	switch(rv) {
		case WAIT_OBJECT_0:
			return PR_SUCCESS;
			break;
		case WAIT_TIMEOUT:
			_PR_THREAD_LOCK(thread);
			PR_ASSERT (thread->state != _PR_IO_WAIT);
			if (thread->wait.cvar != NULL) {
				PR_ASSERT(thread->state == _PR_COND_WAIT);
				thread->wait.cvar = NULL;
				thread->state = _PR_RUNNING;
				_PR_THREAD_UNLOCK(thread);
			} else {
				/* The CVAR was notified just as the timeout
				 * occurred.  This left the semaphore in the
				 * signaled state.  Call WaitForSingleObject()
				 * to clear the semaphore.
				 */
				_PR_THREAD_UNLOCK(thread);
				rv = WaitForSingleObject(thread->md.blocked_sema, INFINITE);
				PR_ASSERT(rv == WAIT_OBJECT_0);
			}
			return PR_SUCCESS;
			break;
		default:
			return PR_FAILURE;
			break;
	}

    return PR_SUCCESS;
}

PRStatus
_PR_MD_WAIT(PRThread *thread, PRIntervalTime ticks)
{
    DWORD rv;

	if (_native_threads_only) {
		return(_native_thread_md_wait(thread, ticks));
	}
    if ( thread->flags & _PR_GLOBAL_SCOPE ) {
        PRUint32 msecs = (ticks == PR_INTERVAL_NO_TIMEOUT) ?
            INFINITE : PR_IntervalToMilliseconds(ticks);
        rv = WaitForSingleObject(thread->md.blocked_sema, msecs);
        switch(rv) {
            case WAIT_OBJECT_0:
                return PR_SUCCESS;
                break;
            case WAIT_TIMEOUT:
                _PR_THREAD_LOCK(thread);
                if (thread->state == _PR_IO_WAIT) {
                    if (thread->io_pending == PR_TRUE) {
                        thread->state = _PR_RUNNING;
                        thread->io_suspended = PR_TRUE;
                        _PR_THREAD_UNLOCK(thread);
                    } else {
                        /* The IO completed just at the same time the timeout
                         * occurred.  This left the semaphore in the signaled
                         * state.  Call WaitForSingleObject() to clear the
                         * semaphore.
                         */
                        _PR_THREAD_UNLOCK(thread);
                        rv = WaitForSingleObject(thread->md.blocked_sema, INFINITE);
                        PR_ASSERT(rv == WAIT_OBJECT_0);
                    }
                } else {
                    if (thread->wait.cvar != NULL) {
                        PR_ASSERT(thread->state == _PR_COND_WAIT);
                        thread->wait.cvar = NULL;
                        thread->state = _PR_RUNNING;
                        _PR_THREAD_UNLOCK(thread);
                    } else {
                        /* The CVAR was notified just as the timeout
                         * occurred.  This left the semaphore in the
                         * signaled state.  Call WaitForSingleObject()
                         * to clear the semaphore.
                         */
                        _PR_THREAD_UNLOCK(thread);
                        rv = WaitForSingleObject(thread->md.blocked_sema, INFINITE);
                        PR_ASSERT(rv == WAIT_OBJECT_0);
                    }
                }
                return PR_SUCCESS;
                break;
            default:
                return PR_FAILURE;
                break;
        }
    } else {
        PRInt32 is;

        _PR_INTSOFF(is);
        _PR_MD_SWITCH_CONTEXT(thread);
    }

    return PR_SUCCESS;
}

static void
_native_thread_io_nowait(
    PRThread *thread,
    int rv,
    int bytes)
{
    int rc;

    PR_ASSERT(rv != 0);
    _PR_THREAD_LOCK(thread);
    if (thread->state == _PR_IO_WAIT) {
        PR_ASSERT(thread->io_suspended == PR_FALSE);
        PR_ASSERT(thread->io_pending == PR_TRUE);
        thread->state = _PR_RUNNING;
        thread->io_pending = PR_FALSE;
        _PR_THREAD_UNLOCK(thread);
    } else {
        /* The IO completed just at the same time the
         * thread was interrupted. This left the semaphore
         * in the signaled state. Call WaitForSingleObject()
         * to clear the semaphore.
         */
        PR_ASSERT(thread->io_suspended == PR_TRUE);
        PR_ASSERT(thread->io_pending == PR_TRUE);
        thread->io_pending = PR_FALSE;
        _PR_THREAD_UNLOCK(thread);
        rc = WaitForSingleObject(thread->md.blocked_sema, INFINITE);
        PR_ASSERT(rc == WAIT_OBJECT_0);
    }

    thread->md.blocked_io_status = rv;
    thread->md.blocked_io_bytes = bytes;
    rc = ResetEvent(thread->md.thr_event);
    PR_ASSERT(rc != 0);
    return;
}

static PRStatus
_native_thread_io_wait(PRThread *thread, PRIntervalTime ticks)
{
    DWORD rv, bytes;
#define _NATIVE_IO_WAIT_HANDLES		2
#define _NATIVE_WAKEUP_EVENT_INDEX	0
#define _NATIVE_IO_EVENT_INDEX		1

	HANDLE wait_handles[_NATIVE_IO_WAIT_HANDLES];

	PRUint32 msecs = (ticks == PR_INTERVAL_NO_TIMEOUT) ?
		INFINITE : PR_IntervalToMilliseconds(ticks);

    PR_ASSERT(thread->flags & _PR_GLOBAL_SCOPE);

	wait_handles[0] = thread->md.blocked_sema;
	wait_handles[1] = thread->md.thr_event;
	rv = WaitForMultipleObjects(_NATIVE_IO_WAIT_HANDLES, wait_handles,
										FALSE, msecs);

	switch(rv) {
		case WAIT_OBJECT_0 + _NATIVE_IO_EVENT_INDEX:
			/*
			 * I/O op completed
			 */
			_PR_THREAD_LOCK(thread);
			if (thread->state == _PR_IO_WAIT) {

				PR_ASSERT(thread->io_suspended == PR_FALSE);
				PR_ASSERT(thread->io_pending == PR_TRUE);
				thread->state = _PR_RUNNING;
				thread->io_pending = PR_FALSE;
				_PR_THREAD_UNLOCK(thread);
			} else {
				/* The IO completed just at the same time the
				 * thread was interrupted. This led to us being
				 * notified twice. Call WaitForSingleObject()
				 * to clear the semaphore.
				 */
				PR_ASSERT(thread->io_suspended == PR_TRUE);
				PR_ASSERT(thread->io_pending == PR_TRUE);
				thread->io_pending = PR_FALSE;
				_PR_THREAD_UNLOCK(thread);
				rv = WaitForSingleObject(thread->md.blocked_sema,
							INFINITE);
				PR_ASSERT(rv == WAIT_OBJECT_0);
			}

			rv = GetOverlappedResult((HANDLE) thread->io_fd,
				&thread->md.overlapped.overlapped, &bytes, FALSE);

			thread->md.blocked_io_status = rv;
			if (rv != 0) {
				thread->md.blocked_io_bytes = bytes;
			} else {
				thread->md.blocked_io_error = GetLastError();
				PR_ASSERT(ERROR_IO_PENDING != thread->md.blocked_io_error);
			}
			rv = ResetEvent(thread->md.thr_event);
			PR_ASSERT(rv != 0);
			break;
		case WAIT_OBJECT_0 + _NATIVE_WAKEUP_EVENT_INDEX:
			/*
			 * I/O interrupted; 
			 */
#ifdef DEBUG
			_PR_THREAD_LOCK(thread);
			PR_ASSERT(thread->io_suspended == PR_TRUE);
			_PR_THREAD_UNLOCK(thread);
#endif
			break;
		case WAIT_TIMEOUT:
			_PR_THREAD_LOCK(thread);
			if (thread->state == _PR_IO_WAIT) {
				thread->state = _PR_RUNNING;
				thread->io_suspended = PR_TRUE;
				_PR_THREAD_UNLOCK(thread);
			} else {
				/*
				 * The thread was interrupted just as the timeout
				 * occurred. This left the semaphore in the signaled
				 * state. Call WaitForSingleObject() to clear the
				 * semaphore.
				 */
				PR_ASSERT(thread->io_suspended == PR_TRUE);
				_PR_THREAD_UNLOCK(thread);
				rv = WaitForSingleObject(thread->md.blocked_sema, INFINITE);
				PR_ASSERT(rv == WAIT_OBJECT_0);
			}
			break;
		default:
			return PR_FAILURE;
			break;
	}

    return PR_SUCCESS;
}


static PRStatus
_NT_IO_WAIT(PRThread *thread, PRIntervalTime timeout)
{
    PRBool fWait = PR_TRUE;

	if (_native_threads_only) {
		return(_native_thread_io_wait(thread, timeout));
	}
    if (!_PR_IS_NATIVE_THREAD(thread))  {

        _PR_THREAD_LOCK(thread);

        /* The IO may have already completed; if so, don't add to sleepQ, 
         * since we are already on the runQ!
         */
        if (thread->io_pending == PR_TRUE) {
            _PR_SLEEPQ_LOCK(thread->cpu);
            _PR_ADD_SLEEPQ(thread, timeout);
            _PR_SLEEPQ_UNLOCK(thread->cpu);
        } else
            fWait = PR_FALSE;
        _PR_THREAD_UNLOCK(thread);
    }
    if (fWait)
        return _PR_MD_WAIT(thread, timeout);
    else
        return PR_SUCCESS;
}

/*
 * Unblock threads waiting for I/O
 * used when interrupting threads
 *
 * NOTE: The thread lock should held when this function is called.
 * On return, the thread lock is released.
 */
void _PR_Unblock_IO_Wait(PRThread *thr)
{
    PRStatus rv;
    _PRCPU *cpu = thr->cpu;
 
    PR_ASSERT(thr->state == _PR_IO_WAIT);
	/*
	 * A thread for which an I/O timed out or was interrupted cannot be
	 * in an IO_WAIT state except as a result of calling PR_Close or
	 * PR_NT_CancelIo for the FD. For these two cases, _PR_IO_WAIT state
	 * is not interruptible
	 */
	if (thr->md.interrupt_disabled == PR_TRUE) {
    	_PR_THREAD_UNLOCK(thr);
		return;
	}
    thr->io_suspended = PR_TRUE;
    thr->state = _PR_RUNNABLE;

    if (!_PR_IS_NATIVE_THREAD(thr)) {
        PRThread *me = _PR_MD_CURRENT_THREAD();
        PR_ASSERT(thr->flags & (_PR_ON_SLEEPQ | _PR_ON_PAUSEQ));
        _PR_SLEEPQ_LOCK(cpu);
        _PR_DEL_SLEEPQ(thr, PR_TRUE);
        _PR_SLEEPQ_UNLOCK(cpu);
		/*
		 * this thread will continue to run on the same cpu until the
		 * I/O is aborted by closing the FD or calling CancelIO
		 */
		thr->md.thr_bound_cpu = cpu;

        PR_ASSERT(!(thr->flags & _PR_IDLE_THREAD));
        _PR_AddThreadToRunQ(me, thr);
    }
    _PR_THREAD_UNLOCK(thr);
    rv = _PR_MD_WAKEUP_WAITER(thr);
    PR_ASSERT(PR_SUCCESS == rv);
}

/* Resume an outstanding IO; requires that after the switch, we disable */
static PRStatus
_NT_ResumeIO(PRThread *thread, PRIntervalTime ticks)
{
    PRBool fWait = PR_TRUE;

    if (!_PR_IS_NATIVE_THREAD(thread)) {
        if (_pr_use_static_tls) {
            _pr_io_restarted_io = thread;
        } else {
            TlsSetValue(_pr_io_restartedIOIndex, thread);
        }
    } else {
        _PR_THREAD_LOCK(thread);
        if (!thread->io_pending)
            fWait = PR_FALSE;
        thread->io_suspended = PR_FALSE;
            
        _PR_THREAD_UNLOCK(thread);
    }
    /* We don't put ourselves back on the sleepQ yet; until we 
     * set the suspended bit to false, we can't do that.  Just save
     * the sleep time here, and then continue.  The restarted_io handler
     * will add us to the sleepQ if needed.
     */
    thread->sleep = ticks;

    if (fWait) {
        if (!_PR_IS_NATIVE_THREAD(thread))
            return _PR_MD_WAIT(thread, ticks);
        else
            return _NT_IO_WAIT(thread, ticks);
    }
    return PR_SUCCESS;
}

PRStatus
_PR_MD_WAKEUP_WAITER(PRThread *thread)
{
    if (thread == NULL) {
        /* If thread is NULL, we aren't waking a thread, we're just poking
         * idle thread 
         */
        if ( PostQueuedCompletionStatus(_pr_completion_port, 0, 
            KEY_CVAR, NULL) == FALSE) 
            return PR_FAILURE;
        return PR_SUCCESS;
    }

    if ( _PR_IS_NATIVE_THREAD(thread) ) {
        if (ReleaseSemaphore(thread->md.blocked_sema, 1, NULL) == FALSE)
            return PR_FAILURE;
        else
            return PR_SUCCESS;
    } else {
        PRThread *me = _PR_MD_CURRENT_THREAD();

        /* When a Native thread has to awaken a user thread, it has to poke
         * the completion port because all user threads might be idle, and
         * thus the CPUs are just waiting for a completion.  
         *