unisock.cpp

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

        // connect UNIX socket
        union { 
            sockaddr sock;
            char     name[MAX_HOST_NAME];
        } u;
        u.sock.sa_family = AF_UNIX;

        assert(strlen(unix_socket_dir) + strlen(address) 
               < MAX_HOST_NAME - offsetof(sockaddr,sa_data)); 
 
        int len = offsetof(sockaddr,sa_data) +
            sprintf(u.sock.sa_data, "%s%s.%u", unix_socket_dir, hostname, port);
        
        while (true) {
            if ((fd = socket(u.sock.sa_family, SOCK_STREAM, 0)) < 0) { 
                errcode = errno;
                TRACE_IMSG(("Failed to create socket: %d\n", errcode));
                return false;
            }
            do { 
                rc = ::connect(fd, &u.sock, len);
            } while (rc < 0 && errno == EINTR);
            
            if (rc < 0) { 
                errcode = errno;
                ::close(fd);
                if (errcode == ENOENT || errcode == ECONNREFUSED) {
                    if (--max_attempts > 0) { 
                        sleep(timeout);
                    } else { 
                        TRACE_IMSG(("All attempts to establish connection are failed\n"));
                        break;
                    }
                } else {
                    TRACE_IMSG(("Failed to establish connection: %d\n", errcode));
                    return false;
                }
            } else {
                errcode = ok;
                state = ss_open;
                return true;
            }
        }
    } else { 
        sockaddr_in sock_inet;
        struct hostent ent;  // entry in hosts table
        struct hostent* hp;
        char buf[GETHOSTBYNAME_BUF_SIZE];
        int h_err;
        if (gethostbyname_r(hostname, &ent, buf, sizeof buf, &hp, &h_err) != 0
            || hp == NULL || hp->h_addrtype != AF_INET)
        {
            TRACE_IMSG(("Host name can not be resolved: %d\n", errno));
            errcode = bad_address;
            return false;
        }
        sock_inet.sin_family = AF_INET;  
        sock_inet.sin_port = htons(port);
        //fprintf(stderr, "Try to connect to '%s' port %d\n", hostname, port);
        
        while (true) {
            for (int i = 0; hp->h_addr_list[i] != NULL; i++) { 
                memcpy(&sock_inet.sin_addr, hp->h_addr_list[i],
                       sizeof sock_inet.sin_addr);
                if ((fd = socket(sock_inet.sin_family, SOCK_STREAM, 0)) < 0) { 
                    errcode = errno;
                    TRACE_IMSG(("Failed to create socket: %d\n", errcode));
                    return false;
                }
                do { 
                    rc = ::connect(fd,(sockaddr*)&sock_inet,sizeof(sock_inet));
                } while (rc < 0 && errno == EINTR);
                
                if (rc < 0) { 
                    errcode = errno;
                    ::close(fd);
                    if (errcode != ENOENT && errcode != ECONNREFUSED) {
                        TRACE_IMSG(("Failed to establish connection: %d\n", errcode));
                        return false;
                    }
                } else {
#if SOCK_NO_DELAY
                    int enabled = 1;
                    if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, 
                                   (char*)&enabled, sizeof enabled) != 0)
                    {
                        errcode = errno;
                        TRACE_IMSG(("Failed to set socket option TCP_NODELAY: %d\n", errcode));
                        ::close(fd);    
                        return false;
                    }
#endif
#if SOCK_LINGER
                    static struct linger l = {1, LINGER_TIME};
                    if (setsockopt(fd, SOL_SOCKET, SO_LINGER, (char*)&l, sizeof l) != 0) { 
                        errcode = errno;
                        TRACE_IMSG(("Failed to set socket option SOL_SOCKET: %d\n", errcode));
                        ::close(fd);
                        return NULL; 
                    }
#endif
                    errcode = ok;
                    state = ss_open;
                    return true;
                }
            }
            if (--max_attempts > 0) { 
                sleep(timeout);
            } else { 
                TRACE_IMSG(("All attempts to establish connection are failed\n"));
                break;
            }
        }
    }
    errcode = connection_failed;
    return false;
}

int unix_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_IMSG(("Socket is not openned\n"));
        return -1;
    }
    if (timeout != WAIT_FOREVER) { 
        start = time(NULL); 
    }
    do { 
        ssize_t rc; 
        if (timeout != WAIT_FOREVER) { 
            fd_set events;
            struct timeval tm;
            FD_ZERO(&events);
            FD_SET(fd, &events);
            tm.tv_sec = timeout;
            tm.tv_usec = 0;
            while ((rc = select(fd+1, &events, NULL, NULL, &tm)) < 0 
                   && errno == EINTR);
            if (rc < 0) { 
                errcode = errno;
                TRACE_IMSG(("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;  
        }
        while ((rc = ::read(fd, (char*)buf + size, max_size - size)) < 0 
               && errno == EINTR); 
        if (rc < 0) { 
            errcode = errno;
            TRACE_IMSG(("Socket read is failed: %d\n", errcode));
            return -1;
        } else if (rc == 0) {
            errcode = broken_pipe;
            TRACE_IMSG(("Socket is disconnected\n"));
            return -1; 
        } else { 
            size += rc; 
        }
    } while (size < min_size); 

    return (int)size;
}

        
bool unix_socket::write(void const* buf, size_t size)
{ 
    if (state != ss_open) { 
        errcode = not_opened;
        TRACE_IMSG(("Socket is not openned\n"));
        return false;
    }
    
    do { 
        ssize_t rc; 
        while ((rc = ::write(fd, buf, size)) < 0 && errno == EINTR); 
        if (rc < 0) { 
            errcode = errno;
            TRACE_IMSG(("Socket write is failed: %d\n", errcode));
            return false;
        } else if (rc == 0) {
            errcode = broken_pipe;
            TRACE_IMSG(("Socket is disconnected\n"));
            return false; 
        } else { 
            buf = (char*)buf + rc; 
            size -= rc; 
        }
    } while (size != 0); 

    //
    // errcode is not assigned 'ok' value beacuse write function 
    // can be called in parallel with other socket operations, so
    // we want to preserve old error code here.
    //
    return true;
}
        
bool unix_socket::close()
{
    if (state != ss_close) {
        state = ss_close;
        if (::close(fd) == 0) {
            errcode = ok;
            return true;
        } else { 
            errcode = errno;
            TRACE_IMSG(("Socket close is failed: %d\n", errcode));
            return false;
        }
    }
    errcode = ok;
    return true;
}

bool unix_socket::shutdown()
{
    if (state == ss_open) { 
        state = ss_shutdown;
        int rc = ::shutdown(fd, 2);
        if (rc != 0) { 
            errcode = errno;
            TRACE_IMSG(("Socket shutdown is failed: %d\n", errcode));
            return false;
        } 
    } 
    return true;
}

unix_socket::~unix_socket()
{
    close();
    if (create_file) { 
        char name[MAX_HOST_NAME];
        char* p = strrchr(address, ':');
        sprintf(name, "%s%.*s.%s", unix_socket_dir, (int)(p - address), address, p+1);
        unlink(name);
    }
    delete[] address;
}

unix_socket::unix_socket(const char* addr, socket_domain domain)
{ 
    address = new char[strlen(addr)+1]; 
    strcpy(address, addr);
    this->domain = domain;
    create_file = false;
    errcode = ok;
}

unix_socket::unix_socket(int new_fd) 
{ 
    fd = new_fd; 
    address = NULL; 
    create_file = false;
    state = ss_open; 
    errcode = ok;
}

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

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

socket_t* socket_t::connect(char const* address, 
                            socket_domain domain, 
                            int max_attempts, 
                            time_t timeout)
{
    unix_socket* sock = new unix_socket(address, domain);
    sock->connect(max_attempts, timeout); 
    return sock;
}

int unix_socket::get_handle()
{
    return fd;
}

END_FASTDB_NAMESPACE