ntio.c

Netscape NSPR库源码

            fd->secret->md.io_model_committed = PR_TRUE;
        }
        /*
         * The accepted socket inherits the nonblocking and
         * inheritable (HANDLE_FLAG_INHERIT) attributes of
         * the listening socket.
         */
        accept_sock = _nt_nonblock_accept(fd, (struct sockaddr *)raddr, rlen, timeout);
        if (!fd->secret->nonblocking) {
            u_long zero = 0;

            rv = ioctlsocket(accept_sock, FIONBIO, &zero);
            PR_ASSERT(0 == rv);
        }
        return accept_sock;
    }

    if (me->io_suspended) {
        PR_SetError(PR_INVALID_STATE_ERROR, 0);
        return -1;
    }

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

    if (!me->md.acceptex_buf) {
        me->md.acceptex_buf = PR_MALLOC(2*INET_ADDR_PADDED);
        if (!me->md.acceptex_buf) {
            PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
            return -1;
        }
    }

    accept_sock = _md_get_recycled_socket();
    if (accept_sock == INVALID_SOCKET)
        return -1;

    memset(&(me->md.overlapped.overlapped), 0, sizeof(OVERLAPPED));
	if (_native_threads_only)
		me->md.overlapped.overlapped.hEvent = me->md.thr_event;

    _PR_THREAD_LOCK(me);
	if (_PR_PENDING_INTERRUPT(me)) {
		me->flags &= ~_PR_INTERRUPT;
		PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
    	_PR_THREAD_UNLOCK(me);
		return -1;
	}
    me->io_pending = PR_TRUE;
    me->state = _PR_IO_WAIT;
    _PR_THREAD_UNLOCK(me);
    me->io_fd = osfd;

    rv = AcceptEx((SOCKET)osfd,
                  accept_sock,
                  me->md.acceptex_buf,
                  0,
                  INET_ADDR_PADDED,
                  INET_ADDR_PADDED,
                  &bytes,
                  &(me->md.overlapped.overlapped));

    if ( (rv == 0) && ((err = GetLastError()) != ERROR_IO_PENDING))  {
        /* Argh! The IO failed */
		_PR_THREAD_LOCK(me);
		me->io_pending = PR_FALSE;
		me->state = _PR_RUNNING;
		if (_PR_PENDING_INTERRUPT(me)) {
			me->flags &= ~_PR_INTERRUPT;
			PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
			_PR_THREAD_UNLOCK(me);
			return -1;
		}
		_PR_THREAD_UNLOCK(me);

		_PR_MD_MAP_ACCEPTEX_ERROR(err);
        return -1;
    }

    if (_native_threads_only && rv) {
        _native_thread_io_nowait(me, rv, bytes);
    } else if (_NT_IO_WAIT(me, timeout) == PR_FAILURE) {
        PR_ASSERT(0);
        return -1;
    }

    PR_ASSERT(me->io_pending == PR_FALSE || me->io_suspended == PR_TRUE);

    if (me->io_suspended) {
        if (_PR_PENDING_INTERRUPT(me)) {
            me->flags &= ~_PR_INTERRUPT;
            PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
        } else {
            PR_SetError(PR_IO_TIMEOUT_ERROR, 0);
        }
        return -1;
    }

    if (me->md.blocked_io_status == 0) {
		_PR_MD_MAP_ACCEPTEX_ERROR(me->md.blocked_io_error);
        return -1;
    }

    if (!fast)
        _PR_MD_UPDATE_ACCEPT_CONTEXT((SOCKET)accept_sock, (SOCKET)osfd);

    /* IO is done */
    GetAcceptExSockaddrs(
            me->md.acceptex_buf,
            0,
            INET_ADDR_PADDED,
            INET_ADDR_PADDED,
            (LPSOCKADDR *)&(Laddr),
            &llen,
            (LPSOCKADDR *)&(Raddr),
            (unsigned int *)rlen);

    if (raddr != NULL)
        memcpy((char *)raddr, (char *)&Raddr->inet, *rlen);

    PR_ASSERT(me->io_pending == PR_FALSE);

    return accept_sock;
}

PRInt32
_PR_MD_FAST_ACCEPT_READ(PRFileDesc *sd, PRInt32 *newSock, PRNetAddr **raddr, 
                   void *buf, PRInt32 amount, PRIntervalTime timeout, 
                   PRBool fast, _PR_AcceptTimeoutCallback callback, 
                   void *callbackArg)
{
    PRInt32 sock = sd->secret->md.osfd;
    PRThread *me = _PR_MD_CURRENT_THREAD();
    int bytes;
    PRNetAddr *Laddr;
    PRUint32 llen, rlen, err;
    int rv;
    PRBool isConnected;
    PRBool madeCallback = PR_FALSE;

    if (me->io_suspended) {
        PR_SetError(PR_INVALID_STATE_ERROR, 0);
        return -1;
    }

    if (!sd->secret->md.io_model_committed) {
        rv = _md_Associate((HANDLE)sock);
        PR_ASSERT(0 != rv);
        sd->secret->md.io_model_committed = PR_TRUE;
    }

    *newSock = _md_get_recycled_socket();
    if (*newSock == INVALID_SOCKET)
        return -1;

    memset(&(me->md.overlapped.overlapped), 0, sizeof(OVERLAPPED));
	if (_native_threads_only)
		me->md.overlapped.overlapped.hEvent = me->md.thr_event;

    _PR_THREAD_LOCK(me);
	if (_PR_PENDING_INTERRUPT(me)) {
		me->flags &= ~_PR_INTERRUPT;
		PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
    	_PR_THREAD_UNLOCK(me);
		return -1;
	}
    me->io_pending = PR_TRUE;
    me->state = _PR_IO_WAIT;
    _PR_THREAD_UNLOCK(me);
    me->io_fd = sock;

    rv = AcceptEx((SOCKET)sock,
                  *newSock,
                  buf,
                  amount,
                  INET_ADDR_PADDED,
                  INET_ADDR_PADDED,
                  &bytes,
                  &(me->md.overlapped.overlapped));

    if ( (rv == 0) && ((err = GetLastError()) != ERROR_IO_PENDING)) {
		_PR_THREAD_LOCK(me);
		me->io_pending = PR_FALSE;
		me->state = _PR_RUNNING;
		if (_PR_PENDING_INTERRUPT(me)) {
			me->flags &= ~_PR_INTERRUPT;
			PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
			_PR_THREAD_UNLOCK(me);
			return -1;
		}
		_PR_THREAD_UNLOCK(me);

		_PR_MD_MAP_ACCEPTEX_ERROR(err);
        return -1;
    }

    if (_native_threads_only && rv) {
        _native_thread_io_nowait(me, rv, bytes);
    } else if (_NT_IO_WAIT(me, timeout) == PR_FAILURE) {
        PR_ASSERT(0);
        return -1;
    }

retry:
    if (me->io_suspended) {
        PRInt32 err;
        INT seconds;
        INT bytes = sizeof(seconds);

        PR_ASSERT(timeout != PR_INTERVAL_NO_TIMEOUT);

        err = getsockopt(*newSock, 
                         SOL_SOCKET,
                         SO_CONNECT_TIME,
                         (char *)&seconds,
                         (PINT)&bytes);
        if ( err == NO_ERROR ) {
            PRIntervalTime elapsed = PR_SecondsToInterval(seconds);

            if (seconds == 0xffffffff) 
                isConnected = PR_FALSE;
            else 
                isConnected = PR_TRUE;

            if (!isConnected) {
                if (madeCallback == PR_FALSE && callback)
                    callback(callbackArg);
                madeCallback = PR_TRUE;
                me->state = _PR_IO_WAIT;
                if (_NT_ResumeIO(me, timeout) == PR_FAILURE)
                    return -1;
                goto retry;
            }

            if (elapsed < timeout) {
                /* Socket is connected but time not elapsed, RESUME IO */
                timeout -= elapsed;
                me->state = _PR_IO_WAIT;
                if (_NT_ResumeIO(me, timeout) == PR_FAILURE)
                    return -1;
                goto retry;
            }
        } else {
            /*  What to do here? Assume socket not open?*/
            PR_ASSERT(0);
            isConnected = PR_FALSE;
        }

        rv = _NT_IO_ABORT(*newSock);

        PR_ASSERT(me->io_pending ==  PR_FALSE);
        PR_ASSERT(me->io_suspended ==  PR_FALSE);
        PR_ASSERT(me->md.thr_bound_cpu ==  NULL);
        /* If the IO is still suspended, it means we didn't get any 
         * completion from NT_IO_WAIT.  This is not disasterous, I hope,
         * but it may mean we still have an IO outstanding...  Try to 
         * recover by just allowing ourselves to continue.
         */
        me->io_suspended = PR_FALSE;
        if (_PR_PENDING_INTERRUPT(me)) {
            me->flags &= ~_PR_INTERRUPT;
            PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
        } else {
            PR_SetError(PR_IO_TIMEOUT_ERROR, 0);
        }
        me->state = _PR_RUNNING;
        return -1;
    }

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

    if (me->md.blocked_io_status == 0) {
		_PR_MD_MAP_ACCEPTEX_ERROR(me->md.blocked_io_error);
        closesocket(*newSock);
        return -1;
    }

    if (!fast) 
        _PR_MD_UPDATE_ACCEPT_CONTEXT((SOCKET)*newSock, (SOCKET)sock);

    /* IO is done */
    GetAcceptExSockaddrs(
            buf,
            amount,
            INET_ADDR_PADDED,
            INET_ADDR_PADDED,
            (LPSOCKADDR *)&(Laddr),
            &llen,
            (LPSOCKADDR *)(raddr),
            (unsigned int *)&rlen);

    return me->md.blocked_io_bytes;
}

PRInt32
_PR_MD_SENDFILE(PRFileDesc *sock, PRSendFileData *sfd,
					PRInt32 flags, PRIntervalTime timeout)
{
    PRThread *me = _PR_MD_CURRENT_THREAD();
    PRInt32 tflags;
    int rv, err;

    if (me->io_suspended) {
        PR_SetError(PR_INVALID_STATE_ERROR, 0);
        return -1;
    }

    if (!sock->secret->md.io_model_committed) {
        rv = _md_Associate((HANDLE)sock->secret->md.osfd);
        PR_ASSERT(0 != rv);
        sock->secret->md.io_model_committed = PR_TRUE;
    }
    if (!me->md.xmit_bufs) {
        me->md.xmit_bufs = PR_NEW(TRANSMIT_FILE_BUFFERS);
        if (!me->md.xmit_bufs) {
            PR_SetError(PR_OUT_OF_MEMORY_ERROR, 0);
            return -1;
        }
    }
    me->md.xmit_bufs->Head       = (void *)sfd->header;
    me->md.xmit_bufs->HeadLength = sfd->hlen;
    me->md.xmit_bufs->Tail       = (void *)sfd->trailer;
    me->md.xmit_bufs->TailLength = sfd->tlen;

    memset(&(me->md.overlapped.overlapped), 0, sizeof(OVERLAPPED));
    me->md.overlapped.overlapped.Offset = sfd->file_offset;
	if (_native_threads_only)
		me->md.overlapped.overlapped.hEvent = me->md.thr_event;

    tflags = 0;
    if (flags & PR_TRANSMITFILE_CLOSE_SOCKET)
        tflags = TF_DISCONNECT | TF_REUSE_SOCKET;

    _PR_THREAD_LOCK(me);
	if (_PR_PENDING_INTERRUPT(me)) {
		me->flags &= ~_PR_INTERRUPT;
		PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
    	_PR_THREAD_UNLOCK(me);
		return -1;
	}
    me->io_pending = PR_TRUE;
    me->state = _PR_IO_WAIT;
    _PR_THREAD_UNLOCK(me);
    me->io_fd = sock->secret->md.osfd;

    rv = TransmitFile((SOCKET)sock->secret->md.osfd,
                      (HANDLE)sfd->fd->secret->md.osfd,
                      (DWORD)sfd->file_nbytes,
                      (DWORD)0,
                      (LPOVERLAPPED)&(me->md.overlapped.overlapped),
                      (TRANSMIT_FILE_BUFFERS *)me->md.xmit_bufs,
                      (DWORD)tflags);
    if ( (rv == 0) && ((err = GetLastError()) != ERROR_IO_PENDING) ) {
		_PR_THREAD_LOCK(me);
		me->io_pending = PR_FALSE;
		me->state = _PR_RUNNING;
		if (_PR_PENDING_INTERRUPT(me)) {
			me->flags &= ~_PR_INTERRUPT;
			PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
			_PR_THREAD_UNLOCK(me);
			return -1;
		}
		_PR_THREAD_UNLOCK(me);

		_PR_MD_MAP_TRANSMITFILE_ERROR(err);
        return -1;
    }

    if (_NT_IO_WAIT(me, timeout) == PR_FAILURE) {
        PR_ASSERT(0);
        return -1;
    }

    PR_ASSERT(me->io_pending == PR_FALSE || me->io_suspended == PR_TRUE);

    if (me->io_suspended) {
        if (_PR_PENDING_INTERRUPT(me)) {
            me->flags &= ~_PR_INTERRUPT;
            PR_SetError(PR_PENDING_INTERRUPT_ERROR, 0);
        } else {
            PR_SetError(PR_IO_TIMEOUT_ERROR, 0);
        }
        return -1;
    }

    if (me->md.blocked_io_status == 0) {
		_PR_MD_MAP_TRANSMITFILE_ERROR(me->md.blocked_io_error);
        return -1;
    }

    if (flags & PR_TRANSMITFILE_CLOSE_SOCKET) {
        _md_put_recycled_socket(sock->secret->md.osfd);
    }

    PR_ASSERT(me->io_pending == PR_FALSE);

    return me->md.blocked_io_bytes;
}

PRInt32
_PR_MD_RECV(PRFileDesc *fd, void *buf, PRInt32 amount, PRIntn flags, 
            PRIntervalTime timeout)
{
    PRInt32 osfd = fd->secret->md.osfd;
    PRThread *me = _PR_MD_CURRENT_THREAD();
    int bytes;
    int rv, err;

    if (_NT_USE_NB_IO(fd)) {
        if (!fd->secret->md.io_model_committed) {
            rv = _md_MakeNonblock((HANDLE)osfd);
            PR_ASSERT(0 != rv);
            fd->secret->md.io_model_committed = PR_TRUE;
        }
        return _nt_nonblock_recv(fd, buf, amount, flags, timeout);
    }

    if (me->io_suspended) {
        PR_SetError(PR_INVALID_STATE_ERROR, 0);
        return -1;
    }