w32sock.cpp

FastDb是高效的内存数据库系统

//-< W32SOCK.CXX >---------------------------------------------------*--------*
// FastDB                    Version 1.0         (c) 1997  GARRET    *     ?  *
// (Main Memory Database Management System)                          *   /\|  *
//                                                                   *  /  \  *
//                          Created:      8-May-97    K.A. Knizhnik  * / [] \ *
//                          Last update: 19-May-97    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Windows sockets  
//-------------------------------------------------------------------*--------*

#define INSIDE_FASTDB

#include "stdtp.h"
#include "w32sock.h"
#include "sync.h"

BEGIN_FASTDB_NAMESPACE

#define MAX_HOST_NAME         256
#define MILLISECOND           1000

static HANDLE WatchDogMutex;

#if !defined(ASM_CPUID_NOT_SUPPORTED) && defined(_MSC_VER) && _MSC_VER < 1200
#define ASM_CPUID_NOT_SUPPORTED
#endif

#ifdef ASM_CPUID_NOT_SUPPORTED
static CRITICAL_SECTION cs;
#endif

class win_socket_library { 
  public:
    SYSTEM_INFO sinfo;
    
    win_socket_library() { 
        WSADATA wsa;
        if (WSAStartup(MAKEWORD(1, 1), &wsa) != 0) {
            fprintf(stderr,"Failed to initialize windows sockets: %d\n",
                    WSAGetLastError());
        }
        //
        // This mutex is used to recognize process termination
        //
        WatchDogMutex = CreateMutex(FASTDB_SECURITY_ATTRIBUTES, TRUE, NULL);

#ifdef PHAR_LAP
        sinfo.wProcessorLevel = 5;
#else
        GetSystemInfo(&sinfo);  
#endif
#ifdef ASM_CPUID_NOT_SUPPORTED
        InitializeCriticalSection(&cs);
#endif
    }
    ~win_socket_library() {
#ifdef ASM_CPUID_NOT_SUPPORTED
        DeleteCriticalSection(&cs);
#endif
        //      WSACleanup();
    }
};

static win_socket_library ws32_lib;

#ifdef __BORLANDC__
static
#else
inline 
#endif
void serialize() { 
#ifndef ASM_CPUID_NOT_SUPPORTED
    if (ws32_lib.sinfo.wProcessorLevel >= 5) { // Pemtium or higher 
#ifdef __MINGW32__
        __asm__ ("CPUID\n" : : );
#else
        __asm CPUID;
#endif
    }
#else
    EnterCriticalSection(&cs);
    LeaveCriticalSection(&cs);
#endif
}

bool win_socket::open(int listen_queue_size)
{
    unsigned short port;
    char* p;
    char hostname[MAX_HOST_NAME];

    assert(address != NULL);

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

    if ((s = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET) { 
        errcode = WSAGetLastError();
        TRACE_MSG(("Socket create is failed: %d\n", errcode));
        return false;
    }
    struct sockaddr_in insock;
    insock.sin_family = AF_INET;
    if (*hostname && stricmp(hostname, "localhost") != 0) {
        struct hostent* hp;  // entry in hosts table
        if ((hp = gethostbyname(hostname)) == NULL 
            || hp->h_addrtype != AF_INET) 
        {
            TRACE_MSG(("Failed to get host by name: %s\n", errno));
            errcode = bad_address;
            return false;
        }
        memcpy(&insock.sin_addr, hp->h_addr, sizeof insock.sin_addr);
    } else {
        insock.sin_addr.s_addr = htonl(INADDR_ANY);
    }
    insock.sin_port = htons(port);

    int on = 1;
    setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (char*)&on, sizeof on);
    
    if (bind(s, (sockaddr*)&insock, sizeof(insock)) != 0) { 
        errcode = WSAGetLastError();
        TRACE_MSG(("Socket bind is failed: %d\n", errcode));
        closesocket(s);
        return false;
    }
    if (listen(s, listen_queue_size) != 0) {
        errcode = WSAGetLastError();
        TRACE_MSG(("Socket listen is failed: %d\n", errcode));
        closesocket(s);
        return false;
    } 
    errcode = ok;
    state = ss_open;
    return true;
}

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

int win_socket::get_handle()
{
    return s;
}

void win_socket::get_error_text(char* buf, size_t buf_size)
{
    char* msg; 
    char  msgbuf[64];

    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: 
#ifndef PHAR_LAP
        {
          int   len;
#ifdef _WINCE
		  wchar_t cnvBuf[CNV_BUF_SIZE];
          FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM,
                              NULL,
                              errcode,
                              0,
                              cnvBuf,
                              CNV_BUF_SIZE-1,
                              NULL);
		  cnvBuf[CNV_BUF_SIZE-1] = '\0';
		  len = wcstombs(buf, cnvBuf, buf_size);
#else
          len = FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM,
                              NULL,
                              errcode,
                              0,
                              buf,
                              buf_size-1,
                              NULL);
#endif
          if (len == 0) { 
              sprintf(msgbuf, "unknown error code %u", errcode);
              msg = msgbuf;
          } else { 
              return;
          }
        }
#else
        sprintf(msgbuf, "System error code: %u", errcode);
        msg = msgbuf;
#endif
    }
    strncpy(buf, msg, buf_size-1);
    buf[buf_size-1] = '\0';
}

socket_t* win_socket::accept()
{
    if (state != ss_open) { 
        errcode = not_opened;
        TRACE_MSG(("Socket not openned\n"));
        return NULL;
    }

    SOCKET new_sock = ::accept(s, NULL, NULL );

    if (new_sock == INVALID_SOCKET) { 
        errcode = WSAGetLastError();
        TRACE_MSG(("Socket accept failed: %d\n", errcode));
        return NULL;
    } else { 
#if SOCK_LINGER
        static struct linger l = {1, LINGER_TIME};
        if (setsockopt(new_sock, SOL_SOCKET, SO_LINGER, (char*)&l, sizeof l) != 0) { 
            errcode = WSAGetLastError();
            TRACE_MSG(("Failed to set socket options: %d\n", errcode));
            closesocket(new_sock);
            return NULL; 
        }
#endif
#if SOCK_NO_DELAY
        int enabled = 1;
        if (setsockopt(new_sock, IPPROTO_TCP, TCP_NODELAY, (char*)&enabled, 
                       sizeof enabled) != 0)
        {
            errcode = WSAGetLastError();
            TRACE_MSG(("Failed to set socket options: %d\n", errcode));
            closesocket(new_sock);      
            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 win_socket(new_sock); 
    }
}

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


bool win_socket::connect(int max_attempts, time_t timeout)
{
    char hostname[MAX_HOST_NAME];
    char *p;
    unsigned short port;

    assert(address != NULL);

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

    struct sockaddr_in insock;  // inet socket address
    struct hostent*    hp;      // entry in hosts table

    if ((hp = gethostbyname(hostname)) == NULL || hp->h_addrtype != AF_INET) {
        TRACE_MSG(("Host name can not be resolved: %d\n", WSAGetLastError()));
        errcode = bad_address;
        return false;
    }
    insock.sin_family = AF_INET;
    insock.sin_port = htons(port);
    
    while (true) {
        for (int i = 0; hp->h_addr_list[i] != NULL; i++) { 
            memcpy(&insock.sin_addr, hp->h_addr_list[i],
                   sizeof insock.sin_addr);
            if ((s = socket(AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET) { 
                errcode = WSAGetLastError();
                TRACE_MSG(("Failed to create socket: %d\n", errcode));
                return false;
            }
            if (::connect(s, (sockaddr*)&insock, sizeof insock) != 0) { 
                errcode = WSAGetLastError();
                closesocket(s);
                if (errcode != WSAECONNREFUSED) {
                    TRACE_MSG(("Failed to establish connection: %d\n", errcode));
                    return false;
                }
            } else {
#if SOCK_NO_DELAY
                int enabled = 1;
                if (setsockopt(s, IPPROTO_TCP, TCP_NODELAY, (char*)&enabled, 
                               sizeof enabled) != 0)
                {
                    errcode = WSAGetLastError();
                    TRACE_MSG(("Failed to set socket options: %d\n", errcode));
                    closesocket(s);     
                    return false;
                }
#endif
#if SOCK_LINGER
                static struct linger l = {1, LINGER_TIME};
                if (setsockopt(s, SOL_SOCKET, SO_LINGER, (char*)&l, sizeof l) != 0) { 
                    errcode = WSAGetLastError();
                    TRACE_MSG(("Failed to set socket options: %d\n", errcode));
                    closesocket(s);
                    return NULL; 
                }
#endif
                errcode = ok;
                state = ss_open;
                return true;
            }
        }
        if (--max_attempts > 0) {  
            Sleep(timeout*MILLISECOND);
        } else { 
            errcode = connection_failed;
            TRACE_MSG(("All attempts to establish connection are failed\n"));
            return false;
        }
    }
}

int win_socket::read(void* buf, size_t min_size, size_t max_size, 
                     time_t timeout)
{ 
    size_t size = 0;
    time_t start = 0;
    if (state != ss_open) { 
        errcode = not_opened;
        TRACE_MSG(("Socket is not openned\n"));
        return -1;
    }
    if (timeout != WAIT_FOREVER) { 
        start = time(NULL); 
    }

    do { 
        int rc;
        if (timeout != WAIT_FOREVER) { 
            fd_set events;
            struct timeval tm;
            FD_ZERO(&events);
            FD_SET(s, &events);
            tm.tv_sec = timeout;
            tm.tv_usec = 0;
            rc = select(s+1, &events, NULL, NULL, &tm);
            if (rc < 0) { 
                errcode = WSAGetLastError();
                TRACE_MSG(("Socket select is failed: %d\n", errcode));
                return -1;
            }
            if (rc == 0) {
                return size;
            }
            time_t now = time(NULL);
            timeout = start + timeout >= now ? timeout + start - now : 0;  
        }
        rc = recv(s, (char*)buf + size, max_size - size, 0);
        if (rc < 0) { 
            errcode = WSAGetLastError();
            TRACE_MSG(("Socket read is failed: %d\n", errcode));
            return -1;
        } else if (rc == 0) {
            errcode = broken_pipe;
            TRACE_MSG(("Socket is disconnected\n"));
            return -1; 
        } else { 
            size += rc; 
        }
    } while (size < min_size); 

    return (int)size;
}
        
bool win_socket::write(void const* buf, size_t size)
{ 
    if (state != ss_open) { 
        errcode = not_opened;
        TRACE_MSG(("Socket is not openned\n"));
        return false;
    }
    
    do { 
        int rc = send(s, (char*)buf, size, 0);
        if (rc < 0) { 
            errcode = WSAGetLastError();
            TRACE_MSG(("Socket write is failed: %d\n", errcode));
            return false;
        } else if (rc == 0) {
            errcode = broken_pipe;
            TRACE_MSG(("Socket is disconnected\n"));
            return false; 
        } else { 
            buf = (char*)buf + rc; 
            size -= rc; 
        }
    } while (size != 0); 

    return true;
}
        
bool win_socket::shutdown()
{
    if (state == ss_open) { 
        state = ss_shutdown;
        int rc = ::shutdown(s, 2);
        if (rc != 0) {
            errcode = WSAGetLastError();
            TRACE_MSG(("Socket shutdown is failed: %d\n", errcode));
            return false;
        } 
    } 
    errcode = ok;
    return true;
}


bool win_socket::close()
{
    if (state != ss_close) { 
        state = ss_close;
        if (closesocket(s) == 0) { 
            errcode = ok;
            return true;
        } else { 
            errcode = WSAGetLastError();
            TRACE_MSG(("Socket close is failed: %d\n", errcode));
            return false;
        }
    }
    return true;
}

char* win_socket::get_peer_name()
{
    if (state != ss_open) { 
        errcode = not_opened;
        return NULL;
    }
    struct sockaddr_in insock;
    int len = sizeof(insock);
    if (getpeername(s, (struct sockaddr*)&insock, &len) != 0) { 
        errcode = WSAGetLastError();
        return NULL;
    }
    char* addr = inet_ntoa(insock.sin_addr);
    if (addr == NULL) { 
        errcode = WSAGetLastError();
        return NULL;
    }
    char* addr_copy = new char[strlen(addr)+1];
    strcpy(addr_copy, addr);
    errcode = ok;
    return addr_copy;
}

win_socket::~win_socket()
{
    close();
    delete[] address;
}

win_socket::win_socket(const char* addr)
{ 
    address = new char[strlen(addr)+1];
    strcpy(address, addr); 
    errcode = ok;
    s = INVALID_SOCKET;
}

win_socket::win_socket(SOCKET new_sock) 
{ 
    s = new_sock; 
    address = NULL; 
    state = ss_open;
    errcode = ok;
}

socket_t* socket_t::create_local(char const* address, int listen_queue_size)
{
#ifdef _WINCE
    return NULL;
#else
    local_win_socket* sock = new local_win_socket(address);
    sock->open(listen_queue_size);
    return sock;
#endif
}

socket_t* socket_t::create_global(char const* address, int listen_queue_size)
{
    win_socket* sock = new win_socket(address);
    sock->open(listen_queue_size); 
    return sock;
}

socket_t* socket_t::connect(char const* address, 
                            socket_domain domain, 
                            int max_attempts,
                            time_t timeout)