ntio.c

Netscape NSPR库源码

         * XXXMB - can we know when we are truely idle (and not checking 
         *         the runq)?
         */
        if ((_PR_IS_NATIVE_THREAD(me) || (thread->cpu != me->cpu)) &&
                (!thread->md.thr_bound_cpu)) {
            /* The thread should not be in any queue */
            PR_ASSERT(thread->queueCount == 0);
            if ( PostQueuedCompletionStatus(_pr_completion_port, 0, 
                KEY_CVAR, &(thread->md.overlapped.overlapped)) == FALSE) 
                return PR_FAILURE;
        }
        return PR_SUCCESS;
    }
}

void
_PR_MD_INIT_IO()
{
    WORD WSAVersion = 0x0101;
    WSADATA WSAData;
    int err;
    OSVERSIONINFO OSversion;

    err = WSAStartup( WSAVersion, &WSAData );
    PR_ASSERT(0 == err);
                                                      
    _pr_completion_port = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 
                                                 NULL, 
                                                 0, 
                                                 0);
 
    _MD_NEW_LOCK(&_pr_recycle_lock);
    _MD_NEW_LOCK(&_pr_ioq_lock);

    OSversion.dwOSVersionInfoSize = sizeof(OSVERSIONINFO);
    if (GetVersionEx(&OSversion)) {
        _nt_version_gets_lockfile_completion = PR_FALSE;
        if (OSversion.dwMajorVersion >= 4) {
            _nt_version_gets_lockfile_completion = PR_TRUE;
        }
    } else 
        PR_ASSERT(0);

    IsFileLocalInit();

    /*
     * UDP support: start up the continuation thread
     */

    pt_tq.op_count = 0;
    pt_tq.head = pt_tq.tail = NULL;
    pt_tq.ml = PR_NewLock();
    PR_ASSERT(NULL != pt_tq.ml);
    pt_tq.new_op = PR_NewCondVar(pt_tq.ml);
    PR_ASSERT(NULL != pt_tq.new_op);
#if defined(DEBUG)
    memset(&pt_debug, 0, sizeof(struct pt_debug_s));
#endif

    pt_tq.thread = PR_CreateThread(
        PR_SYSTEM_THREAD, ContinuationThread, NULL,
        PR_PRIORITY_URGENT, PR_GLOBAL_THREAD, PR_JOINABLE_THREAD, 0);

    PR_ASSERT(NULL != pt_tq.thread);

#ifdef DEBUG
    /* Doublecheck _pr_filetime_offset's hard-coded value is correct. */
    {
        SYSTEMTIME systime;
        union {
           PRTime prt;
           FILETIME ft;
        } filetime;
        BOOL rv;

        systime.wYear = 1970;
        systime.wMonth = 1;
        /* wDayOfWeek is ignored */
        systime.wDay = 1;
        systime.wHour = 0;
        systime.wMinute = 0;
        systime.wSecond = 0;
        systime.wMilliseconds = 0;

        rv = SystemTimeToFileTime(&systime, &filetime.ft);
        PR_ASSERT(0 != rv);
        PR_ASSERT(filetime.prt == _pr_filetime_offset);
    }
#endif /* DEBUG */

    _PR_NT_InitSids();
}

/* --- SOCKET IO --------------------------------------------------------- */

/* _md_get_recycled_socket()
 * Get a socket from the recycle bin; if no sockets are in the bin,
 * create one.  The socket will be passed to AcceptEx() as the
 * second argument.
 */
static SOCKET
_md_get_recycled_socket()
{
    SOCKET rv;
    int af = AF_INET;

    _MD_LOCK(&_pr_recycle_lock);
    if (_pr_recycle_tail) {
        _pr_recycle_tail--;
        rv = _pr_recycle_array[_pr_recycle_tail];
        _MD_UNLOCK(&_pr_recycle_lock);
        return rv;
    }
    _MD_UNLOCK(&_pr_recycle_lock);

    rv = _PR_MD_SOCKET(af, SOCK_STREAM, 0);
    if (rv != INVALID_SOCKET && _md_Associate((HANDLE)rv) == 0) {
        closesocket(rv);
        return INVALID_SOCKET;
    }
    return rv;
}

/* _md_put_recycled_socket()
 * Add a socket to the recycle bin.
 */
static void
_md_put_recycled_socket(SOCKET newsock)
{
    PR_ASSERT(_pr_recycle_tail >= 0);

    _MD_LOCK(&_pr_recycle_lock);
    if (_pr_recycle_tail < RECYCLE_SIZE) {
        _pr_recycle_array[_pr_recycle_tail] = newsock;
        _pr_recycle_tail++;
        _MD_UNLOCK(&_pr_recycle_lock);
    } else {
        _MD_UNLOCK(&_pr_recycle_lock);
        closesocket(newsock);
    }
 
    return;
}

/* _md_Associate()
 * Associates a file with the completion port.
 * Returns 0 on failure, 1 on success.
 */
PRInt32
_md_Associate(HANDLE file)
{
    HANDLE port;

	if (!_native_threads_only) {
		port = CreateIoCompletionPort((HANDLE)file, 
										_pr_completion_port, 
										KEY_IO,
										0);

		/* XXX should map error codes on failures */
		return (port == _pr_completion_port);
	} else {
		return 1;
	}
}

/*
 * _md_MakeNonblock()
 * Make a socket nonblocking.
 * Returns 0 on failure, 1 on success.
 */
static PRInt32
_md_MakeNonblock(HANDLE file)
{
    int rv;
    u_long one = 1;

    rv = ioctlsocket((SOCKET)file, FIONBIO, &one);
    /* XXX should map error codes on failures */
    return (rv == 0);
}

static int missing_completions = 0;
static int max_wait_loops = 0;

static PRInt32
_NT_IO_ABORT(PRInt32 sock)
{
    PRThread *me = _PR_MD_CURRENT_THREAD();
    PRBool fWait;
    PRInt32 rv;
    int loop_count;

    /* This is a clumsy way to abort the IO, but it is all we can do.
     * It looks a bit racy, but we handle all the cases. 
     * case 1:  IO completes before calling closesocket
     *     case 1a:  fWait is set to PR_FALSE
     *           This should e the most likely case.  We'll properly
     *           not wait call _NT_IO_WAIT, since the closesocket()
     *           won't be forcing a completion.
     *     case 1b: fWait is set to PR_TRUE
     *           This hopefully won't happen much.  When it does, this
     *           thread will timeout in _NT_IO_WAIT for CLOSE_INTERVAL
     *           before cleaning up.
     * case 2:  IO does not complete before calling closesocket
     *     case 2a: IO never completes
     *           This is the likely case.  We'll close it and wait
     *           for the completion forced by the close.  Return should
     *           be immediate.
     *     case 2b: IO completes just after calling closesocket
     *           Since the closesocket is issued, we'll either get a
     *           completion back for the real IO or for the close.  We
     *           don't really care.  It may not even be possible to get
     *           a real completion here.  In any event, we'll awaken
     *           from NT_IO_WAIT immediately.
     */

    _PR_THREAD_LOCK(me);
    fWait = me->io_pending;
    if (fWait) {
        /*
         * If there's still I/O pending, it should have already timed
         * out once before this function is called.
         */
        PR_ASSERT(me->io_suspended == PR_TRUE);

        /* Set up to wait for I/O completion again */
        me->state = _PR_IO_WAIT;
        me->io_suspended = PR_FALSE;
        me->md.interrupt_disabled = PR_TRUE;
    }
    _PR_THREAD_UNLOCK(me);

    /* Close the socket if there is one */
    if (sock != INVALID_SOCKET) {
        rv = closesocket((SOCKET)sock);
    }

    /* If there was I/O pending before the close, wait for it to complete */
    if (fWait) {

        /* Wait and wait for the I/O to complete */
        for (loop_count = 0; fWait; ++loop_count) {

            _NT_IO_WAIT(me, CLOSE_TIMEOUT);

            _PR_THREAD_LOCK(me);
            fWait = me->io_pending;
            if (fWait) {
                PR_ASSERT(me->io_suspended == PR_TRUE);
                me->state = _PR_IO_WAIT;
                me->io_suspended = PR_FALSE;
            }
            _PR_THREAD_UNLOCK(me);

            if (loop_count > max_wait_loops) {
                max_wait_loops = loop_count;
            }
        }

        if (loop_count > 1) {
            ++missing_completions;
        }

        me->md.interrupt_disabled = PR_FALSE;
        me->io_pending = PR_FALSE;
        me->state = _PR_RUNNING;
    }

    PR_ASSERT(me->io_pending == PR_FALSE);
    me->md.thr_bound_cpu = NULL;
    me->io_suspended = PR_FALSE;

    return rv;
}


PRInt32
_PR_MD_SOCKET(int af, int type, int flags)
{
    SOCKET sock;

    sock = socket(af, type, flags);

    if (sock == INVALID_SOCKET) {
        _PR_MD_MAP_SOCKET_ERROR(WSAGetLastError());
    }

    return (PRInt32)sock;
}

struct connect_data_s {
    PRInt32 status;
    PRInt32 error;
    PRInt32 osfd;
    struct sockaddr *addr;
    PRUint32 addrlen;
    PRIntervalTime timeout;
};

void
_PR_MD_connect_thread(void *cdata)
{
    struct connect_data_s *cd = (struct connect_data_s *)cdata;

    cd->status = connect(cd->osfd, cd->addr, cd->addrlen);

    if (cd->status == SOCKET_ERROR)
        cd->error = WSAGetLastError();

    return;
}


PRInt32
_PR_MD_CONNECT(PRFileDesc *fd, const PRNetAddr *addr, PRUint32 addrlen, 
               PRIntervalTime timeout)
{
    PRInt32 osfd = fd->secret->md.osfd;
    PRInt32 rv, err;
    u_long nbio;
    PRInt32 rc;

    if (fd->secret->nonblocking) {
        if (!fd->secret->md.io_model_committed) {
            rv = _md_MakeNonblock((HANDLE)osfd);
            PR_ASSERT(0 != rv);
            fd->secret->md.io_model_committed = PR_TRUE;
        }

        if ((rv = connect(osfd, (struct sockaddr *) addr, addrlen)) == -1) {
            err = WSAGetLastError();
            _PR_MD_MAP_CONNECT_ERROR(err);
        }
        return rv;
    }

    /*
     * Temporarily make the socket non-blocking so that we can
     * initiate a non-blocking connect and wait for its completion
     * (with a timeout) in select.
     */
    PR_ASSERT(!fd->secret->md.io_model_committed);
    nbio = 1;
    rv = ioctlsocket((SOCKET)osfd, FIONBIO, &nbio);
    PR_ASSERT(0 == rv);

    rc = _nt_nonblock_connect(fd, (struct sockaddr *) addr, addrlen, timeout);

    /* Set the socket back to blocking. */
    nbio = 0;
    rv = ioctlsocket((SOCKET)osfd, FIONBIO, &nbio);
    PR_ASSERT(0 == rv);

    return rc;
}

PRInt32
_PR_MD_BIND(PRFileDesc *fd, const PRNetAddr *addr, PRUint32 addrlen)
{
    PRInt32 rv;
#if 0
    int one = 1;
#endif

    rv = bind(fd->secret->md.osfd, (const struct sockaddr *)&(addr->inet), addrlen);

    if (rv == SOCKET_ERROR) {
        _PR_MD_MAP_BIND_ERROR(WSAGetLastError());
        return -1;
    }

#if 0
    /* Disable nagle- so far unknown if this is good or not...
     */
    rv = setsockopt(fd->secret->md.osfd, 
                    SOL_SOCKET,
                    TCP_NODELAY,
                    (const char *)&one,
                    sizeof(one));
    PR_ASSERT(rv == 0);
#endif

    return 0;
}

void _PR_MD_UPDATE_ACCEPT_CONTEXT(PRInt32 accept_sock, PRInt32 listen_sock)
{
    /* Sockets accept()'d with AcceptEx need to call this setsockopt before
     * calling anything other than ReadFile(), WriteFile(), send(), recv(), 
     * Transmitfile(), and closesocket().  In order to call any other 
     * winsock functions, we have to make this setsockopt call.
     *
     * XXXMB - For the server, we *NEVER* need this in
     * the "normal" code path.  But now we have to call it.  This is a waste
     * of a system call.  We'd like to only call it before calling the 
     * obscure socket calls, but since we don't know at that point what the
     * original socket was (or even if it is still alive) we can't do it
     * at that point... 
     */
    setsockopt((SOCKET)accept_sock, 
               SOL_SOCKET, 
               SO_UPDATE_ACCEPT_CONTEXT,
               (char *)&listen_sock,
               sizeof(listen_sock));

}

#define INET_ADDR_PADDED (sizeof(PRNetAddr) + 16)
PRInt32
_PR_MD_FAST_ACCEPT(PRFileDesc *fd, PRNetAddr *raddr, PRUint32 *rlen,
              PRIntervalTime timeout, PRBool fast, 
              _PR_AcceptTimeoutCallback callback, void *callbackArg)
{
    PRInt32 osfd = fd->secret->md.osfd;
    PRThread *me = _PR_MD_CURRENT_THREAD();
    SOCKET accept_sock;
    int bytes;
    PRNetAddr *Laddr;
    PRNetAddr *Raddr;
    PRUint32 llen, err;
    int rv;

    if (_NT_USE_NB_IO(fd)) {
        if (!fd->secret->md.io_model_committed) {
            rv = _md_MakeNonblock((HANDLE)osfd);
            PR_ASSERT(0 != rv);