unisock.cpp

最新版本！fastdb是高效的内存数据库系统

//-< UNISOCK.CPP >---------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1999  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:      8-Feb-97    K.A. Knizhnik  * / [] \ *
//                          Last update: 18-May-97    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Unix sockets  
//-------------------------------------------------------------------*--------*

#include "unisock.h"
#undef BYTE_ORDER

#ifdef VXWORKS
#include "fastdbShim.h"
#else
#include <sys/ioctl.h>
#include <fcntl.h>
#include <sys/time.h>
#include <sys/errno.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/utsname.h>
#ifndef HPUX11
#include <sys/select.h>
#endif
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#if defined(_AIX)
#include <strings.h>
#endif  /* _AIX */
#include <stddef.h>
#include <assert.h>
#include <errno.h>

extern "C" {
#include <netdb.h>
}

#include <signal.h>
#endif // VXWORKS

BEGIN_FASTDB_NAMESPACE

const int MAX_HOST_NAME = 256;
const int GETHOSTBYNAME_BUF_SIZE = 1024;

#ifdef VXWORKS
char* unix_socket::unix_socket_dir = "/comp/socket";
#else
char* unix_socket::unix_socket_dir = "/tmp/";
#endif // VXWORKS

class unix_socket_library { 
  public: 
    unix_socket_library() { 
        static struct sigaction sigpipe_ignore; 
        sigpipe_ignore.sa_handler = SIG_IGN;
        sigaction(SIGPIPE, &sigpipe_ignore, NULL);
    }
};

static unix_socket_library unisock_lib;

bool unix_socket::open(int listen_queue_size)
{
    char hostname[MAX_HOST_NAME];
    unsigned short port;
    char* p;
#ifdef VXWORKS
    int proto = SOCK_STREAM;
#endif // VXWORKS

    assert(address != NULL);

    if ((p = strchr(address, ':')) == NULL 
        || unsigned(p - address) >= sizeof(hostname) 
        || sscanf(p+1, "%hu", &port) != 1) 
    {
        TRACE_IMSG(("Invalid address: %s\n", address));
        errcode = bad_address;
        return false;
    }
    memcpy(hostname, address, p - address);
    hostname[p - address] = '\0';
    
    create_file = false; 
    union { 
        sockaddr    sock;
        sockaddr_in sock_inet;
#ifdef VXWORKS
        struct sockaddr_un usock;
#endif
        char        name[MAX_HOST_NAME];
    } u;
    int len;

    if (domain == sock_local_domain) { 
#ifdef VXWORKS
        memset(&u.usock, 0, sizeof(struct sockaddr_un));
        u.usock.sun_family = AF_UNIX;
        proto = SOCK_SEQPACKET;
        u.usock.sun_len = len = sizeof (struct sockaddr_un);
        sprintf(u.usock.sun_path, "%s/0x%x", unix_socket_dir, port);
        TRACE_IMSG(("Sock %s %d\n", u.usock.sun_path,u.usock.sun_len));
        unlink(u.usock.sun_path); // remove file if existed
        create_file = true;
#else
        u.sock.sa_family = AF_UNIX;

        assert(strlen(unix_socket_dir) + strlen(address) 
               < MAX_HOST_NAME - offsetof(sockaddr,sa_data)); 
        
        len = offsetof(sockaddr,sa_data) + 
            sprintf(u.sock.sa_data, "%s%s.%u", unix_socket_dir, hostname, port);

        unlink(u.sock.sa_data); // remove file if existed
        create_file = true; 
#endif // VXWORKS
    } else {
        u.sock_inet.sin_family = AF_INET;
        if (*hostname && strcmp(hostname, "localhost") != 0) {
            struct hostent* hp;
#if defined(HAVE_GETHOSTBYNAME_R) && !defined(NO_PTHREADS)
            struct hostent ent;  // entry in hosts table
            char buf[GETHOSTBYNAME_BUF_SIZE];
            int h_err;
#if defined(__sun)
            if ((hp = gethostbyname_r(hostname, &ent, buf, sizeof buf, &h_err)) == NULL
#else
            if (gethostbyname_r(hostname, &ent, buf, sizeof buf, &hp, &h_err) != 0
                || hp == NULL
#endif
                || hp->h_addrtype != AF_INET)
#else
            if ((hp = gethostbyname(hostname)) == NULL || hp->h_addrtype != AF_INET) 
#endif
            {
                TRACE_IMSG(("Failed to get host by name: %s\n", errno));
                errcode = bad_address;
                return false;
            }
            memcpy(&u.sock_inet.sin_addr, hp->h_addr, 
                   sizeof u.sock_inet.sin_addr);
        } else {
            u.sock_inet.sin_addr.s_addr = htonl(INADDR_ANY);
        }
        u.sock_inet.sin_port = htons(port);
        len = sizeof(sockaddr_in);      
    } 

#ifdef VXWORKS
    if ((fd = socket(u.sock.sa_family, proto, 0)) < 0) {
        errcode = errno;
        TRACE_IMSG(("Socket create is failed: %d", errcode));
        return false;
    }
#else
    if ((fd = socket(u.sock.sa_family, SOCK_STREAM, 0)) < 0) { 
        errcode = errno;
        TRACE_IMSG(("Socket create is failed: %d\n", errcode));
        return false;
    }
#endif // VXWORKS
    int on = 1;
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof on);
#ifdef VXWORKS
    if (bind(fd, reinterpret_cast<struct sockaddr*>(&u.usock), len) < 0) {
        errcode = errno;
        TRACE_IMSG(("Socket bind is failed: %d", errcode));
        ::close(fd);
        return false;
    }
#else
    if (bind(fd, &u.sock, len) < 0) {
        errcode = errno;
        TRACE_IMSG(("Socket bind is failed: %d\n", errcode));
        ::close(fd);
        return false;
    }
#endif // VXWORKS
    if (listen(fd, listen_queue_size) < 0) {
        errcode = errno;
        TRACE_IMSG(("Socket listen is failed: %d\n", errcode));
        ::close(fd);
        return false;
    }
    errcode = ok;
    state = ss_open;
    return true;
}

char* unix_socket::get_peer_name()
{
    if (state != ss_open) { 
        errcode = not_opened;
        return NULL;
    }
    struct sockaddr_in insock;
    #if defined(__linux__) || (defined(__FreeBSD__) && __FreeBSD__ > 3) || defined(_AIX43) || defined(__OpenBSD__) || defined(HPUX11) || defined(_SOCKLEN_T)
    socklen_t len = sizeof(insock);
#elif defined(_AIX41)
    size_t len = sizeof(insock);
#else
    int len = sizeof(insock);
#endif
    if (getpeername(fd, (struct sockaddr*)&insock, &len) != 0) { 
        errcode = errno;
        return NULL;
    }
    char* addr = inet_ntoa(insock.sin_addr);
    if (addr == NULL) { 
        errcode = errno;
        return NULL;
    }
    char* addr_copy = new char[strlen(addr)+1];
    strcpy(addr_copy, addr);
    errcode = ok;
    return addr_copy;
}

bool  unix_socket::is_ok()
{
    return errcode == ok;
}

void unix_socket::get_error_text(char* buf, size_t buf_size)
{
    char* msg; 
    switch(errcode) { 
      case ok:
        msg = "ok";
        break;
      case not_opened:
        msg = "socket not opened";
        break;
      case bad_address: 
        msg = "bad address";
        break;
      case connection_failed: 
        msg = "exceed limit of attempts of connection to server";
        break;
      case broken_pipe:
        msg = "connection is broken";
        break; 
      case invalid_access_mode:
        msg = "invalid access mode";
        break;
      default: 
        msg = strerror(errcode);
    }
    strncpy(buf, msg, buf_size-1);
    buf[buf_size-1] = '\0';
}

socket_t* unix_socket::accept()
{
    int s;

    if (state != ss_open) { 
        errcode = not_opened;
        TRACE_IMSG(("Socket not openned\n"));
        return NULL;
    }

    while((s = ::accept(fd, NULL, NULL )) < 0 && errno == EINTR);

    if (s < 0) { 
        errcode = errno;
        TRACE_IMSG(("Socket accept failed: %d\n", errcode));
        return NULL;
    } else if (state != ss_open) {
        errcode = not_opened;
        TRACE_IMSG(("Socket not openned\n"));
        return NULL;
    } else { 
#if SOCK_NO_DELAY
        if (domain == sock_global_domain) { 
            int enabled = 1;
            if (setsockopt(s, IPPROTO_TCP, TCP_NODELAY, (char*)&enabled, 
                           sizeof enabled) != 0)
            {
                errcode = errno;
                TRACE_IMSG(("Failed to set socket options: %d\n", errcode));
                ::close(s);     
                return NULL;
            }
        }
#endif
#if SOCK_LINGER
        static struct linger l = {1, LINGER_TIME};
        if (setsockopt(s, SOL_SOCKET, SO_LINGER, (char*)&l, sizeof l) != 0) { 
            TRACE_IMSG(("Failed to set socket options: %d\n", errno));
            errcode = invalid_access_mode; 
            ::close(s);
            return NULL; 
        }
#endif
#if SOCK_SNDBUF_SIZE 
        int size = SOCK_SNDBUF_SIZE;
        setsockopt(s, SOL_SOCKET, SO_SNDBUF, (char*)&size, sizeof size);
#endif
        errcode = ok;
        return new unix_socket(s); 
    }
}

bool unix_socket::cancel_accept() 
{
    bool result = close();
    // Wakeup listener
    delete socket_t::connect(address, domain, 1, 0);
    return result;
}    


bool unix_socket::connect(int max_attempts, time_t timeout)
{
    int   rc;
    char* p;
    struct utsname local_host;
    char hostname[MAX_HOST_NAME];
    unsigned short port;
#ifdef VXWORKS
    int proto = SOCK_STREAM;
#endif // VXWORKS

    assert(address != NULL);

    if ((p = strchr(address, ':')) == NULL 
        || unsigned(p - address) >= sizeof(hostname) 
        || sscanf(p+1, "%hu", &port) != 1) 
    {
        errcode = bad_address;
        TRACE_IMSG(("Invalid address: %s\n", address));
        return false;
    }
    memcpy(hostname, address, p - address);
    hostname[p - address] = '\0';
    
    create_file = false; 
    uname(&local_host);