w32sock.cpp

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

            return false;
        }       
    }
    sprintf(buf, "%s.shr", Name);
    BufHnd = CreateFileMapping(INVALID_HANDLE_VALUE, FASTDB_SECURITY_ATTRIBUTES, PAGE_READWRITE,
                               0, sizeof(socket_buf)*2, W32_STRING(buf));
    if (!BufHnd) {
        Error = GetLastError();
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        return false;
    }
    RcvBuf = (socket_buf*)MapViewOfFile(BufHnd, FILE_MAP_ALL_ACCESS, 0, 0, 0);
    if (!RcvBuf) {
        Error = GetLastError();
        CloseHandle(BufHnd);
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        return false;
    }   
    SndBuf = RcvBuf+1;
    RcvBuf->DataBeg = RcvBuf->DataEnd = 0;
    SndBuf->DataBeg = SndBuf->DataEnd = 0;       
    Error = ok;
    state = ss_open;
    return true;
}

local_win_socket::local_win_socket()
{
    int i;
    BufHnd = NULL;
    Mutex = NULL; 
    Name = NULL;

    for (i = RD; i <= RTT; i++) {  
        Signal[i] = CreateEvent(FASTDB_SECURITY_ATTRIBUTES, false, false, NULL);
        if (!Signal[i]) {
            Error = GetLastError();
            while (--i >= 0) { 
                CloseHandle(Signal[i]);
            }
            return;
        }       
    }
    // create anonymous shared memory section
    BufHnd = CreateFileMapping(INVALID_HANDLE_VALUE, FASTDB_SECURITY_ATTRIBUTES, PAGE_READWRITE,
                               0, sizeof(socket_buf)*2, NULL);
    if (!BufHnd) {
        Error = GetLastError();
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        return;
    }
    RcvBuf = (socket_buf*)MapViewOfFile(BufHnd, FILE_MAP_ALL_ACCESS, 0, 0, 0);
    if (!RcvBuf) {
        Error = GetLastError();
        CloseHandle(BufHnd);
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        BufHnd = NULL;
        return;
    }   
    SndBuf = RcvBuf+1;
    RcvBuf->DataBeg = RcvBuf->DataEnd = 0;
    SndBuf->DataBeg = SndBuf->DataEnd = 0;       
    Error = ok;
    state = ss_open;
}

local_win_socket::~local_win_socket()
{
    close();
    delete[] Name;
}       

socket_t* local_win_socket::accept()
{   
    HANDLE h[2];

    if (state != ss_open) {     
        return NULL;
    }
                    
    connect_data* cdp = (connect_data*)SndBuf->Data;
    cdp->Pid = GetCurrentProcessId();
    cdp->Mutex = WatchDogMutex;
    while (true) { 
        SetEvent(Signal[RTR]);
        int rc = WaitForSingleObject(Signal[RD], ACCEPT_TIMEOUT);
        if (rc == WAIT_OBJECT_0) {
            if (state != ss_open) { 
                Error = not_opened;
                return NULL;
            }
            Error = ok;
            break;
        } else if (rc != WAIT_TIMEOUT) { 
            Error = GetLastError();
            return NULL;
        }
    }
    local_win_socket* sock = new local_win_socket();
    sock->Mutex = ((connect_data*)RcvBuf->Data)->Mutex;
    accept_data* adp = (accept_data*)SndBuf->Data;
    adp->BufHnd = sock->BufHnd;
    for (int i = RD; i <= RTT; i++) { 
        adp->Signal[(i + TD - RD) & RTT] = sock->Signal[i]; 
    }
    SetEvent(Signal[TD]);
    h[0] = Signal[RD];
    h[1] = sock->Mutex;
    int rc = WaitForMultipleObjects(2, h, false, INFINITE);
    if (rc != WAIT_OBJECT_0) {
        if (rc == WAIT_OBJECT_0+1 || rc == WAIT_ABANDONED+1) { 
            Error = broken_pipe;
            ReleaseMutex(Mutex);
        } else { 
            Error = GetLastError();
        }       
        delete sock;
        return NULL;
    }    
    return sock;
}

bool local_win_socket::cancel_accept() 
{
    state = ss_shutdown;
    SetEvent(Signal[RD]);
    SetEvent(Signal[RTT]);
    return true;
}    

char* local_win_socket::get_peer_name()
{
    if (state != ss_open) { 
        Error = not_opened;
        return NULL;
    }
    char* addr = "127.0.0.1";
    char* addr_copy = new char[strlen(addr)+1];
    strcpy(addr_copy, addr);
    Error = ok;
    return addr_copy;
}

bool local_win_socket::is_ok()
{
    return !Error;
}

bool local_win_socket::close()
{
    if (state != ss_close) {            
        state = ss_close;
        if (Mutex) { 
            CloseHandle(Mutex);
        }
        for (int i = RD; i <= RTT; i++) { 
            CloseHandle(Signal[i]);
        }
        UnmapViewOfFile(RcvBuf < SndBuf ? RcvBuf : SndBuf);
        CloseHandle(BufHnd);    
        Error = not_opened;
    }
    return true;
}

void local_win_socket::get_error_text(char* buf, size_t buf_size)
{
    switch (Error) { 
      case ok:
        strncpy(buf, "ok", buf_size-1);
        break;
      case not_opened:
        strncpy(buf, "socket not opened", buf_size-1);
        break;
      case broken_pipe:
        strncpy(buf, "connection is broken", buf_size-1);
        break;
      case timeout_expired:
        strncpy(buf, "connection timeout expired", buf_size-1);
        break;
      default:  
#ifndef PHAR_LAP
        FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM,
                      NULL,
                      Error,
                      0,
                      W32_STRING(buf),
                      buf_size-1,
                      NULL);
#else
        strncpy(buf, "Unknown socket error", buf_size-1);
#endif
    }
    buf[buf_size-1] = '\0';
}


bool local_win_socket::shutdown()
{
    if (state == ss_open) { 
        state = ss_shutdown;
        SetEvent(Signal[RD]);   
        SetEvent(Signal[RTT]);  
    }
    return true;
}

bool local_win_socket::connect(int max_attempts, time_t timeout)
{
    char buf[MAX_ADDRESS_LEN];
    int  rc, i, error_code;
    HANDLE h[2];

    for (i = RD; i <= RTT; i++) {  
        sprintf(buf, "%s.%c", Name, ((i + TD - RD) & RTT) + '0');
        Signal[i] = CreateEvent(FASTDB_SECURITY_ATTRIBUTES, false, false, W32_STRING(buf));
        if (!Signal[i]) {
            Error = GetLastError();
            while (--i >= 0) { 
                CloseHandle(Signal[i]);
            }
            return false;
        }       
    }
    sprintf(buf, "%s.shr", Name);
    BufHnd = CreateFileMapping(INVALID_HANDLE_VALUE, FASTDB_SECURITY_ATTRIBUTES, PAGE_READWRITE,
                               0, sizeof(socket_buf)*2, W32_STRING(buf));
    if (!BufHnd) {
        Error = GetLastError();
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        return false;
    }
    SndBuf = (socket_buf*)MapViewOfFile(BufHnd, FILE_MAP_ALL_ACCESS, 0, 0, 0);
    if (!SndBuf) { 
        Error = GetLastError();
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        CloseHandle(BufHnd);
        return false;
    }
    RcvBuf = SndBuf+1;
    state = ss_shutdown;
    Mutex = NULL;

    rc = WaitForSingleObject(Signal[RTT], (DWORD)(timeout*max_attempts*MILLISECOND));
    if (rc != WAIT_OBJECT_0) {
        error_code = rc == WAIT_TIMEOUT ? timeout_expired : GetLastError();
        close();
        Error = error_code;
        return false;
    }
    connect_data* cdp = (connect_data*)RcvBuf->Data;
    HANDLE hServer = OpenProcess(STANDARD_RIGHTS_REQUIRED|PROCESS_DUP_HANDLE,
                                 false, cdp->Pid);
    if (!hServer) { 
        error_code = GetLastError();
        close();
        Error = error_code;
        return false;
    }
    HANDLE hSelf = GetCurrentProcess();
    if (!DuplicateHandle(hServer, cdp->Mutex, hSelf, &Mutex, 
                         0, false, DUPLICATE_SAME_ACCESS) ||
        !DuplicateHandle(hSelf, WatchDogMutex, hServer, 
                         &((connect_data*)SndBuf->Data)->Mutex, 
                         0, false, DUPLICATE_SAME_ACCESS))
    {
        error_code = GetLastError();
        CloseHandle(hServer);
        close();
        Error = error_code;
        return false;
    }
    SetEvent(Signal[TD]);
    h[0] = Signal[RD];
    h[1] = Mutex;
    rc = WaitForMultipleObjects(2, h, false, INFINITE);

    if (rc != WAIT_OBJECT_0) { 
        if (rc == WAIT_OBJECT_0+1 || rc == WAIT_ABANDONED+1) { 
            error_code = broken_pipe;
            ReleaseMutex(Mutex);
        } else { 
            error_code = GetLastError();
        }
        CloseHandle(hServer);
        close();
        Error = error_code;
        return false;
    }
    accept_data ad = *(accept_data*)RcvBuf->Data;

    SetEvent(Signal[TD]);
    for (i = RD; i <= RTT; i++) { 
        CloseHandle(Signal[i]);
    }
    UnmapViewOfFile(SndBuf);
    CloseHandle(BufHnd);        
    BufHnd = NULL;

    if (!DuplicateHandle(hServer, ad.BufHnd, hSelf, &BufHnd, 
                         0, false, DUPLICATE_SAME_ACCESS))
    {
        Error = GetLastError();
        CloseHandle(hServer);
        CloseHandle(Mutex); 
        return false;
    } else { 
        for (i = RD; i <= RTT; i++) { 
            if (!DuplicateHandle(hServer, ad.Signal[i], 
                                 hSelf, &Signal[i], 
                                 0, false, DUPLICATE_SAME_ACCESS))
            {
                Error = GetLastError();
                CloseHandle(hServer);
                CloseHandle(BufHnd); 
                CloseHandle(Mutex); 
                while (--i >= 0) CloseHandle(Signal[1]);
                return false;
            }
        }
    }
    CloseHandle(hServer);

    SndBuf = (socket_buf*)MapViewOfFile(BufHnd, FILE_MAP_ALL_ACCESS, 0, 0, 0);
    if (!SndBuf) { 
        Error = GetLastError();
        CloseHandle(BufHnd); 
        CloseHandle(Mutex); 
        for (i = RD; i <= RTT; i++) {  
            CloseHandle(Signal[i]);
        }
        return false;
    }
    RcvBuf = SndBuf+1;
    Error = ok;
    state = ss_open; 
    return true;
}

int local_win_socket::get_handle()
{
    return -1;
}

#endif

END_FASTDB_NAMESPACE