ncbi_socket.c

ncbi源码

                        if (polls[i].sock->type != eSOCK_Datagram
                            &&  (polls[i].sock->r_status == eIO_Closed  ||
                                 polls[i].sock->eof))
                            break;
                        write_only = 0;
                        FD_SET(fd, &r_fds);
                        if (polls[i].sock->type == eSOCK_Datagram  ||
                            polls[i].event != eIO_Read             ||
                            polls[i].sock->w_status == eIO_Closed  ||
                            n == 1  ||  (!polls[i].sock->pending  &&
                                         !polls[i].sock->w_len))
                            break;
                        read_only = 0;
                        FD_SET(fd, &w_fds);
                        break;
                    default:
                        /* should never get here */
                        assert(0);
                        break;
                    }
                    FD_SET(fd, &e_fds);
                    if (n_fds < (int) fd)
                        n_fds = (int) fd;
                } else {
                    polls[i].revent = eIO_Close;
                    ready = 1;
                }
            } else {
                polls[i].revent = eIO_Close;
                bad = 1;
            }
        }

        if ( bad )
            return eIO_InvalidArg;

        if ( ready )
            return eIO_Success;

        if (!tv  ||  s_Less(s_SelectTimeout, &x_tv)) {
            if ( s_SelectTimeout ) {
                xx_tv = *s_SelectTimeout;
            }
        } else
            xx_tv = x_tv;

        n_fds = select(SOCK_NFDS((TSOCK_Handle) n_fds),
                       write_only ? 0 : &r_fds, read_only ? 0 : &w_fds,
                       &e_fds, tv || s_SelectTimeout ? &xx_tv : 0);

        /* timeout has expired */
        if (n_fds == 0) {
            if ( !tv )
                continue;
            if ( s_Less(s_SelectTimeout, &x_tv) ) {
                x_tv.tv_sec  -= s_SelectTimeout->tv_sec;
                x_tv.tv_usec -= s_SelectTimeout->tv_usec;
                continue;
            }
            return eIO_Timeout;
        }

        if (n_fds > 0)
            break;

        /* n_fds < 0 */
        if (SOCK_ERRNO != SOCK_EINTR) {
            int  x_errno = SOCK_ERRNO;
            char _id[32];
            CORE_LOGF_ERRNO_EX(eLOG_Trace, x_errno, SOCK_STRERROR(x_errno),
                               ("%s[SOCK::s_Select]  Failed select()",
                                n == 1 ? s_ID(polls[0].sock, _id) : ""));
            return eIO_Unknown;
        }

        if ((n != 1  &&  s_InterruptOnSignal == eOn)  ||
            (n == 1  &&  (polls[0].sock->i_on_sig == eOn
                          ||  (polls[0].sock->i_on_sig == eDefault
                               &&  s_InterruptOnSignal == eOn)))) {
            return eIO_Interrupt;
        }
    }

    n_fds = 0;
    for (i = 0; i < n; i++) {
        if ( polls[i].sock ) {
            TSOCK_Handle fd = polls[i].sock->sock;
            assert(polls[i].revent == eIO_Open);
            if (fd != SOCK_INVALID) {
                if (!write_only  &&  FD_ISSET(fd, &r_fds))
                    polls[i].revent = eIO_Read;
                if (!read_only   &&   FD_ISSET(fd, &w_fds))
                    polls[i].revent = (EIO_Event)(polls[i].revent | eIO_Write);
                if (!polls[i].revent  &&  FD_ISSET(fd, &e_fds))
                    polls[i].revent = eIO_Close;
            } else
                polls[i].revent = eIO_Close;
            if (polls[i].revent != eIO_Open)
                n_fds++;
        }
    }
    assert(n_fds != 0);

    /* success; can do I/O now */
    return eIO_Success;
}



/******************************************************************************
 *  UTILITY
 */

extern const STimeout* SOCK_SetSelectInternalRestartTimeout(const STimeout* t)
{
    static struct timeval  s_NewTmo;
    static STimeout        s_OldTmo;
    const  STimeout* retval = s_tv2to(s_SelectTimeout, &s_OldTmo);
    s_SelectTimeout         = s_to2tv(t,               &s_NewTmo);
    return retval;
}



/******************************************************************************
 *  LISTENING SOCKET
 */

extern EIO_Status LSOCK_Create(unsigned short port,
                               unsigned short backlog,
                               LSOCK*         lsock)
{
    return LSOCK_CreateEx(port, backlog, lsock, eDefault);
}


extern EIO_Status LSOCK_CreateEx(unsigned short port,
                                 unsigned short backlog,
                                 LSOCK*         lsock,
                                 ESwitch        log)
{
    unsigned int       x_id = ++s_ID_Counter;
    TSOCK_Handle       x_lsock;
    struct sockaddr_in addr;

    *lsock = 0;
    /* initialize internals */
    verify(s_Initialized  ||  SOCK_InitializeAPI() == eIO_Success);

    /* create new(listening) socket */
    if ((x_lsock = socket(AF_INET, SOCK_STREAM, 0)) == SOCK_INVALID) {
        int x_errno = SOCK_ERRNO;
        CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                           ("LSOCK#%u[?]: [LSOCK::Create] "
                            " Cannot create socket", x_id));
        return eIO_Unknown;
    }

    /*
     * It was confirmed(?) that at least under Solaris 2.5 this precaution:
     * 1) makes the address released immediately after the process
     *    termination;
     * 2) still issue EADDINUSE error on the attempt to bind() to the
     *    same address being in-use by a living process (if SOCK_STREAM).
     */
    if ( !s_SetReuseAddress(x_lsock, 1/*true*/) ) {
        int x_errno = SOCK_ERRNO;
        CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                           ("LSOCK#%u[%u]: [LSOCK::Create] "
                            " Failed setsockopt(REUSEADDR)",
                            x_id, (unsigned int) x_lsock));
        SOCK_CLOSE(x_lsock);
        return eIO_Unknown;
    }

    /* bind */
    memset(&addr, 0, sizeof(addr));
    addr.sin_family      = AF_INET;
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    addr.sin_port        = htons(port);
#ifdef HAVE_SIN_LEN
    addr.sin_len         = sizeof(addr);
#endif /*HAVE_SIN_LEN*/
    if (bind(x_lsock, (struct sockaddr*) &addr, sizeof(addr)) != 0) {
        int x_errno = SOCK_ERRNO;
        CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                           ("LSOCK#%u[%u]: [LSOCK::Create]  Failed bind(:%hu)",
                            x_id, (unsigned int) x_lsock, port));
        SOCK_CLOSE(x_lsock);
        return x_errno == SOCK_EADDRINUSE ? eIO_Closed : eIO_Unknown;
    }

    /* listen */
    if (listen(x_lsock, backlog) != 0) {
        int x_errno = SOCK_ERRNO;
        CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                           ("LSOCK#%u[%u]: [LSOCK::Create]  Failed listen(%hu)"
                            , x_id, (unsigned int) x_lsock, backlog));
        SOCK_CLOSE(x_lsock);
        return eIO_Unknown;
    }

    /* set to non-blocking mode */
    if ( !s_SetNonblock(x_lsock, 1/*true*/) ) {
        CORE_LOGF(eLOG_Error, ("LSOCK#%u[%u]: [LSOCK::Create] "
                               " Cannot set socket to non-blocking mode",
                               x_id, (unsigned int) x_lsock));
        SOCK_CLOSE(x_lsock);
        return eIO_Unknown;
    }

    /* allocate memory for the internal socket structure */
    if ( !(*lsock = (LSOCK) calloc(1, sizeof(**lsock))) )
        return eIO_Unknown;
    (*lsock)->sock     = x_lsock;
    (*lsock)->id       = x_id;
    (*lsock)->log      = log;
    (*lsock)->i_on_sig = eDefault;
    SET_LISTENING(*lsock);

    /* statistics & logging */
    if (log == eOn  ||  (log == eDefault  &&  s_Log == eOn)) {
        CORE_LOGF(eLOG_Trace, ("LSOCK#%u[%u]: Listening at port :%hu",
                               x_id, (unsigned int) x_lsock, port));
    }

    return eIO_Success;
}


extern EIO_Status LSOCK_Accept(LSOCK           lsock,
                               const STimeout* timeout,
                               SOCK*           sock)
{
    struct sockaddr_in addr;
    unsigned int       x_id;
    TSOCK_Handle       x_sock;

    if (lsock->sock == SOCK_INVALID) {
        CORE_LOGF(eLOG_Error, ("LSOCK#%u[?]: [LSOCK::Accept] "
                               " Invalid socket", lsock->id));
        assert(0);
        return eIO_Unknown;
    }

    {{ /* wait for the connection request to come (up to timeout) */
        EIO_Status     status;
        SSOCK_Poll     poll;
        struct timeval tv;

        poll.sock   = (SOCK) lsock;
        poll.event  = eIO_Read;
        poll.revent = eIO_Open;
        if ((status = s_Select(1, &poll, s_to2tv(timeout,&tv))) != eIO_Success)
            return status;
        if (poll.revent == eIO_Close)
            return eIO_Unknown;
        assert(poll.event == eIO_Read  &&  poll.revent == eIO_Read);
    }}

    x_id = (lsock->id * 1000 + ++s_ID_Counter) * 1000;
    {{ /* accept next connection */
        SOCK_socklen_t addrlen = (SOCK_socklen_t) sizeof(addr);
        memset(&addr, 0, sizeof(addr));
#ifdef HAVE_SIN_LEN
        addr.sin_len = sizeof(addr);
#endif
        if ((x_sock = accept(lsock->sock, (struct sockaddr*) &addr, &addrlen))
            == SOCK_INVALID) {
            int x_errno = SOCK_ERRNO;
            CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                               ("LSOCK#%u[%u]: [LSOCK::Accept] "
                                " Failed accept()", lsock->id,
                                (unsigned int) lsock->sock));
            return eIO_Unknown;
        }
        lsock->n_accept++;
        assert(addr.sin_family == AF_INET);
        /* man accept(2) notes that non-blocking state may not be inherited */
        if ( !s_SetNonblock(x_sock, 1/*true*/) ) {
            CORE_LOGF(eLOG_Error, ("SOCK#%u[%u]: [LSOCK::Accept]  Cannot"
                                   " set accepted socket to non-blocking mode",
                                   x_id, (unsigned int) x_sock));
            SOCK_CLOSE(x_sock);
            return eIO_Unknown;
        }

        if (s_ReuseAddress  &&  !s_SetReuseAddress(x_sock, 1/*true*/)) {
            int x_errno = SOCK_ERRNO;
            CORE_LOGF_ERRNO_EX(eLOG_Warning, x_errno, SOCK_STRERROR(x_errno),
                               ("SOCK#%u[%u]: [LSOCK::Accept] "
                                " Failed setsockopt(REUSEADDR)",
                                x_id, (unsigned int) x_sock));
        }
    }}

    /* create new SOCK structure */
    if ( !(*sock = (SOCK) calloc(1, sizeof(**sock))) ) {
        SOCK_CLOSE(x_sock);
        return eIO_Unknown;
    }

    /* success */
    (*sock)->sock     = x_sock;
    (*sock)->id       = x_id;
    (*sock)->host     = addr.sin_addr.s_addr;
    (*sock)->port     = addr.sin_port;
    (*sock)->log      = lsock->log;
    (*sock)->type     = eSOCK_ServerSide;
    (*sock)->r_on_w   = eDefault;
    (*sock)->i_on_sig = eDefault;
    (*sock)->r_status = eIO_Success;
    (*sock)->eof      = 0/*false*/;
    (*sock)->w_status = eIO_Success;
    (*sock)->pending  = 0/*connected*/;
    /* all timeouts zeroed - infinite */
    BUF_SetChunkSize(&(*sock)->r_buf, SOCK_BUF_CHUNK_SIZE);
    /* w_buf is unused for accepted sockets */

    /* statistics & logging */
    if (lsock->log == eOn  ||  (lsock->log == eDefault  &&  s_Log == eOn))
        s_DoLog(*sock, eIO_Open, 0, 0, (struct sockaddr*) &addr);

    return eIO_Success;
}


extern EIO_Status LSOCK_Close(LSOCK lsock)
{
    EIO_Status status;

    if (lsock->sock == SOCK_INVALID) {
        CORE_LOGF(eLOG_Error, ("LSOCK#%u[?]: [LSOCK::Close] "
                               " Invalid socket", lsock->id));
        assert(0);
        return eIO_Unknown;
    }

    /* set the socket back to blocking mode */
    if ( !s_SetNonblock(lsock->sock, 0/*false*/) ) {
        CORE_LOGF(eLOG_Warning, ("LSOCK#%u[%u]: [LSOCK::Close] "
                                 " Cannot set socket back to blocking mode",
                                 lsock->id, (unsigned int) lsock->sock));
    }

    /* statistics & logging */
    if (lsock->log == eOn  ||  (lsock->log == eDefault  &&  s_Log == eOn)) {
        CORE_LOGF(eLOG_Trace, ("LSOCK#%u[%u]: Closing at port :%hu "
                               "(%u accept%s total)", lsock->id,
                               (unsigned int) lsock->sock, lsock->port,
                               lsock->n_accept, lsock->n_accept == 1? "":"s"));
    }

    status = eIO_Success;
    for (;;) { /* close persistently - retry if interrupted by a signal */
        /* success */
        if (SOCK_CLOSE(lsock->sock) == 0)
            break;

        /* error */
        if (SOCK_ERRNO != SOCK_EINTR) {
            int x_errno = SOCK_ERRNO;
            CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),