ncbi_socket.c

ncbi源码

                               ("LSOCK#%u[%u]: [LSOCK::Close]  Failed close()",
                                lsock->id, (unsigned int) lsock->sock));
            status = eIO_Unknown;
            break;
        }
    }

    /* cleanup & return */
    lsock->sock = SOCK_INVALID;
    free(lsock);
    return status;
}


extern EIO_Status LSOCK_GetOSHandle(LSOCK  lsock,
                                    void*  handle,
                                    size_t handle_size)
{
    if (!handle  ||  handle_size != sizeof(lsock->sock)) {
        CORE_LOGF(eLOG_Error, ("LSOCK#%u[%u]: [LSOCK::GetOSHandle] "
                               " Invalid handle %s%lu", lsock->id,
                               (unsigned int) lsock->sock, handle? "size ": "",
                               handle ? (unsigned long) handle_size : 0));
        assert(0);
        return eIO_InvalidArg;
    }

    memcpy(handle, &lsock->sock, handle_size);
    return lsock->sock == SOCK_INVALID ? eIO_Closed : eIO_Success;
}



/******************************************************************************
 *  SOCKET
 */

/* connect() could be async/interrupted by a signal or just cannot be
 * established immediately;  yet, it must have been in progress
 * (asynchronous), so wait here for it to succeed (become writable).
 */
static EIO_Status s_IsConnected(SOCK                  sock,
                                const struct timeval* tv,
                                int*                  x_errno,
                                int/*bool*/           writeable)
{
    EIO_Status     status;
#if defined(NCBI_OS_UNIX) || defined(NCBI_OS_MSWIN)
    SOCK_socklen_t x_len = (SOCK_socklen_t) sizeof(*x_errno);
#endif /*NCBI_OS_UNIX || NCBI_OS_MSWIN*/
    SSOCK_Poll     poll;

    *x_errno = 0;
    if (sock->w_status == eIO_Closed)
        return eIO_Closed;
    if ( !writeable ) {
        poll.sock   = sock;
        poll.event  = eIO_Write;
        poll.revent = eIO_Open;
        status      = s_Select(1, &poll, tv);
        if (status == eIO_Timeout)
            return status;
    } else {
        status      = eIO_Success;
        poll.revent = eIO_Write;
    }
#if defined(NCBI_OS_UNIX) || defined(NCBI_OS_MSWIN)
    if (status == eIO_Success  &&
        (getsockopt(sock->sock, SOL_SOCKET, SO_ERROR, (void*) x_errno, &x_len)
         ||  *x_errno != 0)) {
        status = eIO_Unknown;
    }
#endif /*NCBI_OS_UNIX || NCBI_OS_MSWIN*/
    if (status != eIO_Success  ||  poll.revent != eIO_Write) {
        if ( !*x_errno )
            *x_errno = SOCK_ERRNO;
        if (*x_errno == SOCK_ECONNREFUSED)
            sock->r_status = sock->w_status = status = eIO_Closed;
        else if (status == eIO_Success)
            status = eIO_Unknown;
    } else if (s_ReuseAddress  &&  !s_SetReuseAddress(sock->sock, 1/*true*/)) {
        int x_errno = SOCK_ERRNO;
        char _id[32];
        CORE_LOGF_ERRNO_EX(eLOG_Warning, x_errno, SOCK_STRERROR(x_errno),
                           ("%s[SOCK::s_IsConnected]  Failed "
                            "setsockopt(REUSEADDR)", s_ID(sock, _id)));
    }
    return status;
}


/* Connect the (pre-allocated) socket to the specified "host:port" peer.
 * HINT: if "host" is NULL then assume(!) that the "sock" already exists,
 *       and connect to the same host;  the same is for zero "port".
 * NOTE: Client-side stream sockets only.
 */
static EIO_Status s_Connect(SOCK            sock,
                            const char*     host,
                            unsigned short  port,
                            const STimeout* timeout)
{
    char               _id[32];
    int                x_errno;
    TSOCK_Handle       x_sock;
    unsigned int       x_host;
    unsigned short     x_port;
    struct sockaddr_in peer;
    int                n;

    assert(sock->type == eSOCK_ClientSide);
#ifdef NCBI_OS_UNIX
    assert(!sock->file[0]);
#endif /*NCBI_OS_UNIX*/

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

    /* get address of the remote host (assume the same host if it is NULL) */
    x_host = host  &&  *host ? SOCK_gethostbyname(host) : sock->host;
    if ( !x_host ) {
        CORE_LOGF(eLOG_Error, ("%s[SOCK::s_Connect]  Failed "
                               "SOCK_gethostbyname(\"%.64s\")",
                               s_ID(sock, _id), host));
        return eIO_Unknown;
    }

    /* set the port to connect to (assume the same port if "port" is zero) */
    x_port = (unsigned short) (port ? htons(port) : sock->port);

    /* create new socket */
    if ((x_sock = 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),
                           ("%s[SOCK::s_Connect]  Cannot create socket",
                            s_ID(sock, _id)));
        return eIO_Unknown;
    }
    sock->sock = x_sock;

    /* set the socket I/O to non-blocking mode */
    if ( !s_SetNonblock(x_sock, 1/*true*/) ) {
        CORE_LOGF(eLOG_Error, ("%s[SOCK::s_Connect]  Cannot set socket to "
                               "non-blocking mode", s_ID(sock, _id)));
        sock->sock = SOCK_INVALID;
        SOCK_CLOSE(x_sock);
        return eIO_Unknown;
    }

    /* fill in the server "addr" to connect to */
    memset(&peer, 0, sizeof(peer));
    peer.sin_family      = AF_INET;
    peer.sin_addr.s_addr = x_host;
    peer.sin_port        = x_port;
#ifdef HAVE_SIN_LEN
    peer.sin_len         = sizeof(peer);
#endif /*HAVE_SIN_LEN*/

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

    /* establish connection to the peer */
    sock->r_status  = eIO_Success;
    sock->eof       = 0/*false*/;
    sock->w_status  = eIO_Success;
    assert(sock->w_len == 0);
    for (n = 0; ; n = 1) {
        if (connect(x_sock, (struct sockaddr*) &peer, sizeof(peer)) == 0) {
            x_errno = 0;
            break;
        }

        x_errno = SOCK_ERRNO;
        if (x_errno != SOCK_EINTR  ||  sock->i_on_sig == eOn  ||
            (sock->i_on_sig == eDefault  &&  s_InterruptOnSignal))
            break;
    }
    if (x_errno) {
        if ((n != 0 || x_errno != SOCK_EINPROGRESS)  &&
            (n == 0 || x_errno != SOCK_EALREADY)     &&
            x_errno != SOCK_EWOULDBLOCK) {
            if (x_errno != SOCK_EINTR) {
                char addr[80];
                HostPortToString(x_host, ntohs(x_port), addr, sizeof(addr));
                CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                                   ("%s[SOCK::s_Connect]  Failed connect() "
                                    "to %s", s_ID(sock, _id), addr));
            }
            sock->sock = SOCK_INVALID;
            SOCK_CLOSE(x_sock);
            /* unrecoverable error */
            return x_errno == SOCK_EINTR ? eIO_Interrupt : eIO_Unknown;
        }

        if (!timeout  ||  timeout->sec  ||  timeout->usec) {
            EIO_Status     status;
            struct timeval tv;

            status = s_IsConnected(sock, s_to2tv(timeout, &tv), &x_errno, 0);
            if (status != eIO_Success) {
                char addr[80];
                HostPortToString(x_host, ntohs(x_port), addr, sizeof(addr));
                CORE_LOGF_ERRNO_EX(eLOG_Error, x_errno, SOCK_STRERROR(x_errno),
                                   ("%s[SOCK::s_Connect]  Failed pending "
                                    "connect() to %s (%s)", s_ID(sock, _id),
                                    addr, IO_StatusStr(status)));
                sock->sock = SOCK_INVALID;
                SOCK_CLOSE(x_sock);
                return status;
            }
            sock->pending = 0/*connected*/;
        } else
            sock->pending = 1/*not yet connected*/;
    } else
        sock->pending = 0/*connected*/;

    /* success: do not change any timeouts */
    sock->host  = x_host;
    sock->port  = x_port;
    sock->w_len = BUF_Size(sock->w_buf);
    return eIO_Success;
}


/* To allow emulating "peek" using the NCBI data buffering.
 * (MSG_PEEK is not implemented on Mac, and it is poorly implemented
 * on Win32, so we had to implement this feature by ourselves.)
 * NOTE: This call is for stream sockets only.
 */
static int s_Recv(SOCK        sock,
                  void*       buffer,
                  size_t      size,
                  int/*bool*/ peek)
{
    char*  x_buffer = (char*) buffer;
    char   xx_buffer[4096];
    size_t n_read;

    assert(sock->type != eSOCK_Datagram  &&  !sock->pending);
    if ( !size ) {
        /* internal upread use only */
        assert(sock->r_status != eIO_Closed && !sock->eof && peek && !buffer);
        n_read = 0;
    } else {
        /* read (or peek) from the internal buffer */
        n_read = peek ?
            BUF_Peek(sock->r_buf, x_buffer, size) :
            BUF_Read(sock->r_buf, x_buffer, size);
        if ((n_read  &&  (n_read == size  ||  !peek))  ||
            sock->r_status == eIO_Closed  ||  sock->eof) {
            return (int) n_read;
        }
    }

    /* read (not just peek) from the socket */
    do {
        size_t n_todo;
        int    x_read;

        if ( !size ) {
            /* internal upread call -- read out as much as possible */
            n_todo    = sizeof(xx_buffer);
            x_buffer  = xx_buffer;
        } else if ( !buffer ) {
            /* read to the temporary buffer (to store or discard later) */
            n_todo    = size - n_read;
            if (n_todo > sizeof(xx_buffer))
                n_todo = sizeof(xx_buffer);
            x_buffer  = xx_buffer;
        } else {
            /* read to the data buffer provided by user */
            n_todo    = size - n_read;
            x_buffer += n_read;
        }
        /* recv */
        x_read = recv(sock->sock, x_buffer, n_todo, 0);

        /* success */
        if (x_read >= 0  ||
            (x_read < 0  &&  (SOCK_ERRNO == SOCK_ENOTCONN      ||
                              SOCK_ERRNO == SOCK_ECONNRESET    ||
                              SOCK_ERRNO == SOCK_ECONNABORTED  ||
                              SOCK_ERRNO == SOCK_ENETRESET))) {
            /* statistics & logging */
            if (sock->log == eOn  ||  (sock->log == eDefault && s_Log == eOn)){
                s_DoLog(sock, eIO_Read, x_read >= 0 ? x_buffer : 0,
                        (size_t)(x_read < 0 ? 0 : x_read), 0);
            }
            if (x_read <= 0) {
                /* catch EOF/failure */
                sock->eof = 1/*true*/;
                if (x_read == 0)
                    sock->r_status = eIO_Success;
                else
                    sock->r_status = sock->w_status = eIO_Closed;
                break;
            }
        } else {
            /* some error */
            int x_errno = SOCK_ERRNO;
            if (x_errno != SOCK_EWOULDBLOCK  &&
                x_errno != SOCK_EAGAIN       &&
                x_errno != SOCK_EINTR) {
                /* catch unknown ERROR */
                sock->r_status = eIO_Unknown;
                CORE_LOGF_ERRNO_EX(eLOG_Trace, x_errno, SOCK_STRERROR(x_errno),
                                   ("%s[SOCK::s_Recv] "
                                    " Failed recv()", s_ID(sock, xx_buffer)));
            }
            return n_read ? (int) n_read : -1;
        }
        assert(x_read > 0);

        /* if "peek" -- store the new read data in the internal input buffer */
        if (peek  &&  !BUF_Write(&sock->r_buf, x_buffer, (size_t) x_read)) {
            CORE_LOGF_ERRNO(eLOG_Error, errno,
                            ("%s[SOCK::s_Recv]  Cannot store data in "
                             "peek buffer", s_ID(sock, xx_buffer)));
            sock->eof      = 1/*failure*/;
            sock->r_status = eIO_Closed;
            break;
        }

        /* successful read */
        sock->r_status = eIO_Success;
        sock->n_read  += x_read;
        n_read        += x_read;
    } while (!size  ||  (!buffer  &&  n_read < size));

    return (int) n_read;
}


static EIO_Status s_WritePending(SOCK, const struct timeval*, int);

/* s_Select() with stall protection: try pull incoming data from sockets.
 * This method returns array of polls, "revent"s of which are always
 * compatible with requested "event"s. That is, it always strips additional
 * events that s_Select() may have set to indicate additional I/O events
 * some sockets are ready for. Return eIO_Timeout if no compatible events
 * were found (all sockets are not ready for inquired respective I/O) within
 * the specified timeout (and no other socket error was flagged).
 * Return eIO_Success if at least one socket is ready. Return the number
 * of sockets that are ready via pointer argument "n_ready" (may be NULL).
 * Return other error code to indicate error condition.
 */
static EIO_Status s_SelectStallsafe(size_t                n,
                                    SSOCK_Poll            polls[],
                                    const struct timeval* tv,
                                    size_t*               n_ready)
{
    int/*bool*/ pending;
    EIO_Status  status;
    size_t      i, j;

    if ((status = s_Select(n, polls, tv)) != eIO_Success) {
        if ( n_ready )
            *n_ready = 0;
        return status;
    }

    j = 0;
    pe