w32sock.cpp

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

        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 = (long)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, time_t timeout)
{ 
    time_t start = 0;
    if (state != ss_open) { 
        errcode = not_opened;
        TRACE_MSG(("Socket is not openned\n"));
        return false;
    }
    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 = (long)timeout;
            tm.tv_usec = 0;
            rc = select(s+1, NULL, &events, NULL, &tm);
            if (rc <= 0) { 
                errcode = WSAGetLastError();
                TRACE_MSG(("Socket select is failed: %d\n", errcode));
                return false;
            }
            time_t now = time(NULL);
            timeout = start + timeout >= now ? timeout + start - now : 0;  
        }
        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)
{
#ifndef _WINCE
    char   hostname[MAX_HOST_NAME];
    size_t hostname_len;
    char const* port;
    
    if (domain == sock_local_domain 
        || (domain == sock_any_domain 
            && ((port = strchr(address, ':')) == NULL 
                || ((hostname_len = port - address) == 9 
                    && strncmp(address, "localhost", hostname_len) == 0)
                || (gethostname(hostname, sizeof hostname) == 0 
                    && strlen(hostname) == hostname_len 
                    && strncmp(address, hostname, hostname_len) == 0))))
    {
        local_win_socket* s = new local_win_socket(address);
        s->connect(max_attempts, timeout); 
        return s;
    } 
    else 
#endif
    { 
        win_socket* s = new win_socket(address);
        s->connect(max_attempts, timeout); 
        return s;
    }  
}
    
#ifndef _WINCE

//
// Local windows sockets
//

int local_win_socket::read(void* buf, size_t min_size, size_t max_size, 
                           time_t timeout)
{
    time_t start = 0;
    char* dst = (char*)buf;
    size_t size = 0;
    Error = ok;
    if (timeout != WAIT_FOREVER) { 
        start = time(NULL); 
        timeout *= 1000; // convert seconds to miliseconds
    }
    while (size < min_size && state == ss_open) {       
        RcvBuf->RcvWaitFlag = true;
        serialize();
        size_t begin = RcvBuf->DataBeg;
        size_t end = RcvBuf->DataEnd;
        size_t rcv_size = (begin <= end)
            ? end - begin : sizeof(RcvBuf->Data) - begin;
        if (rcv_size > 0) { 
            RcvBuf->RcvWaitFlag = false;
            if (rcv_size >= max_size) { 
                memcpy(dst, &RcvBuf->Data[begin], max_size);
                begin += max_size;
                size += max_size;
            } else { 
                memcpy(dst, &RcvBuf->Data[begin], rcv_size);
                begin += rcv_size;
                dst += rcv_size;
                size += rcv_size;
                max_size -= rcv_size;
            } 
            RcvBuf->DataBeg = (begin == sizeof(RcvBuf->Data)) ? 0 : begin;
            if (RcvBuf->SndWaitFlag) { 
                SetEvent(Signal[RTR]);
            }           
        } else {
            HANDLE h[2];
            h[0] = Signal[RD];
            h[1] = Mutex;
            int rc = WaitForMultipleObjects(2, h, false, (DWORD)timeout);
            RcvBuf->RcvWaitFlag = false;
            if (rc != WAIT_OBJECT_0) {
                if (rc == WAIT_OBJECT_0+1 || rc == WAIT_ABANDONED+1) { 
                    Error = broken_pipe;
                    ReleaseMutex(Mutex);
                } else if (rc == WAIT_TIMEOUT) { 
                    return size;
                } else { 
                    Error = GetLastError();
                }
                return -1;
            }
            if (timeout != WAIT_FOREVER) { 
                time_t now = time(NULL);
                timeout = timeout >= (now - start)*1000 
                    ? timeout - (now - start)*1000 : 0;  
            }
        }
    }                   
    return size < min_size ? -1 : (int)size;
}


bool local_win_socket::write(const void* buf, size_t size, time_t timeout)
{
    char* src = (char*)buf;
    Error = ok;
    while (size > 0 && state == ss_open) {      
        SndBuf->SndWaitFlag = true;
        serialize();
        size_t begin = SndBuf->DataBeg;
        size_t end = SndBuf->DataEnd;
        size_t snd_size = (begin <= end) 
            ? sizeof(SndBuf->Data) - end - (begin == 0)
            : begin - end - 1;
        if (snd_size > 0) { 
            SndBuf->SndWaitFlag = false;
            if (snd_size >= size) { 
                memcpy(&SndBuf->Data[end], src, size);
                end += size;
                size = 0;
            } else { 
                memcpy(&SndBuf->Data[end], src, snd_size);
                end += snd_size;
                src += snd_size;
                size -= snd_size;
            } 
            SndBuf->DataEnd = (end == sizeof(SndBuf->Data)) ? 0 : end;
            if (SndBuf->RcvWaitFlag) { 
                SetEvent(Signal[TD]);
            }           
        } else {
            HANDLE h[2];
            h[0] = Signal[RTT];
            h[1] = Mutex;
            int rc = WaitForMultipleObjects(2, h, false, (DWORD)timeout);
            SndBuf->SndWaitFlag = false;
            if (rc != WAIT_OBJECT_0) {
                if (rc == WAIT_OBJECT_0+1 || rc == WAIT_ABANDONED+1) { 
                    Error = broken_pipe;
                    ReleaseMutex(Mutex);
                } else { 
                    Error = GetLastError();
                }       
                return false;
            }
        }
    }                           
    return size == 0;
}

#define MAX_ADDRESS_LEN 64

local_win_socket::local_win_socket(const char* address)
{
    Name = new char[strlen(address)+1];
    strcpy(Name, address);
    Error = not_opened;
    Mutex = NULL;
}
 
bool local_win_socket::open(int)
{
    char buf[MAX_ADDRESS_LEN];  
    int  i;

    for (i = RD; i <= RTT; i++) {  
        sprintf(buf, "%s.%c", Name, i + '0');
        Signal[i] = CreateEvent(FASTDB_SECURITY_ATTRIBUTES, false, false, W32_STRING(buf));
        if (GetLastError() == ERROR_ALREADY_EXISTS) { 
            WaitForSingleObject(Signal[i], 0);
        }
        if (!Signal[i]) {
            Error = GetLastError();
            while (--i >= 0) { 
                CloseHandle(Signal[i]);
            }