X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=lib%2Fmemcached_connect.c;h=50c6fa62acacd3e40823909343586735cce5d093;hb=85a0b3bc944db78c070961cd3a48b8e34dcef8a4;hp=42390a9af30cd92319c2a8695b9b24c1de5c0c21;hpb=acca3c7ad60716bf073ae8de0652517542cdd224;p=awesomized%2Flibmemcached diff --git a/lib/memcached_connect.c b/lib/memcached_connect.c index 42390a9a..50c6fa62 100644 --- a/lib/memcached_connect.c +++ b/lib/memcached_connect.c @@ -10,9 +10,18 @@ static memcached_return set_hostinfo(memcached_server_st *server) sprintf(str_port, "%u", server->port); memset(&hints, 0, sizeof(hints)); + hints.ai_family= AF_INET; - hints.ai_socktype= SOCK_STREAM; - hints.ai_protocol= 0; + if (server->type == MEMCACHED_CONNECTION_UDP) + { + hints.ai_protocol= IPPROTO_UDP; + hints.ai_socktype= SOCK_DGRAM; + } + else + { + hints.ai_socktype= SOCK_STREAM; + hints.ai_protocol= IPPROTO_TCP; + } e= getaddrinfo(server->hostname, str_port, &hints, &ai); if (e != 0) @@ -29,14 +38,86 @@ static memcached_return set_hostinfo(memcached_server_st *server) return MEMCACHED_SUCCESS; } -static memcached_return unix_socket_connect(memcached_st *ptr, unsigned int server_key) +static memcached_return set_socket_options(memcached_server_st *ptr) +{ + if (ptr->type == MEMCACHED_CONNECTION_UDP) + return MEMCACHED_SUCCESS; + + if (ptr->root->flags & MEM_NO_BLOCK) + { + int error; + struct linger linger; + struct timeval waittime; + + waittime.tv_sec= 10; + waittime.tv_usec= 0; + + linger.l_onoff= 1; + linger.l_linger= MEMCACHED_DEFAULT_TIMEOUT; + error= setsockopt(ptr->fd, SOL_SOCKET, SO_LINGER, + &linger, (socklen_t)sizeof(struct linger)); + WATCHPOINT_ASSERT(error == 0); + + error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDTIMEO, + &waittime, (socklen_t)sizeof(struct timeval)); + WATCHPOINT_ASSERT(error == 0); + + error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVTIMEO, + &waittime, (socklen_t)sizeof(struct timeval)); + WATCHPOINT_ASSERT(error == 0); + } + + if (ptr->root->flags & MEM_TCP_NODELAY) + { + int flag= 1; + int error; + + error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_NODELAY, + &flag, (socklen_t)sizeof(int)); + WATCHPOINT_ASSERT(error == 0); + } + + if (ptr->root->send_size) + { + int error; + + error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF, + &ptr->root->send_size, (socklen_t)sizeof(int)); + WATCHPOINT_ASSERT(error == 0); + } + + if (ptr->root->recv_size) + { + int error; + + error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF, + &ptr->root->recv_size, (socklen_t)sizeof(int)); + WATCHPOINT_ASSERT(error == 0); + } + + /* For the moment, not getting a nonblocking mode will not be fatal */ + if (ptr->root->flags & MEM_NO_BLOCK) + { + int flags; + + flags= fcntl(ptr->fd, F_GETFL, 0); + if (flags != -1) + { + (void)fcntl(ptr->fd, F_SETFL, flags | O_NONBLOCK); + } + } + + return MEMCACHED_SUCCESS; +} + +static memcached_return unix_socket_connect(memcached_server_st *ptr) { struct sockaddr_un servAddr; socklen_t addrlen; - if (ptr->hosts[server_key].fd == -1) + if (ptr->fd == -1) { - if ((ptr->hosts[server_key].fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0) + if ((ptr->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0) { ptr->cached_errno= errno; return MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE; @@ -44,19 +125,59 @@ static memcached_return unix_socket_connect(memcached_st *ptr, unsigned int serv memset(&servAddr, 0, sizeof (struct sockaddr_un)); servAddr.sun_family= AF_UNIX; - strcpy(servAddr.sun_path, ptr->hosts[server_key].hostname); /* Copy filename */ + strcpy(servAddr.sun_path, ptr->hostname); /* Copy filename */ addrlen= strlen(servAddr.sun_path) + sizeof(servAddr.sun_family); test_connect: - if (connect(ptr->hosts[server_key].fd, + if (connect(ptr->fd, (struct sockaddr *)&servAddr, sizeof(servAddr)) < 0) { switch (errno) { + case EINPROGRESS: + { + struct timeval tm = { ptr->root->connect_timeout, 0 }; + socklen_t len= sizeof(int); + fd_set wset; + int error=0, value; + + FD_ZERO(&wset); + FD_SET(ptr->fd, &wset); + + select(ptr->fd+1, NULL, &wset, NULL, &tm); + if (FD_ISSET(ptr->fd, &wset) != 0) + { + if (getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &value, &len) == 0) + { + if (value) + { + error= 1; + } + } + else + { + error= 1; + } + } + else + { + error= 1; + } + + if (error) + { + ptr->cached_errno= errno; + WATCHPOINT_ERRNO(ptr->cached_errno); + close(ptr->fd); + ptr->fd= -1; + return MEMCACHED_ERRNO; + } + + break; + } /* We are spinning waiting on connect */ case EALREADY: - case EINPROGRESS: case EINTR: goto test_connect; case EISCONN: /* We were spinning waiting on connect */ @@ -66,198 +187,104 @@ test_connect: ptr->cached_errno= errno; return MEMCACHED_ERRNO; } - ptr->connected++; - } - } - return MEMCACHED_SUCCESS; -} - -static memcached_return udp_connect(memcached_st *ptr, unsigned int server_key) -{ - if (ptr->hosts[server_key].fd == -1) - { - /* Old connection junk still is in the structure */ - WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0); - - /* - If we have not allocated the hosts object. - Or if the cache has not been set. - */ - if (ptr->hosts[server_key].sockaddr_inited == MEMCACHED_NOT_ALLOCATED || - (!(ptr->flags & MEM_USE_CACHE_LOOKUPS))) - { - memcached_return rc; - - rc= set_hostinfo(&ptr->hosts[server_key]); - if (rc != MEMCACHED_SUCCESS) - return rc; - - ptr->hosts[server_key].sockaddr_inited= MEMCACHED_ALLOCATED; - } - - /* Create the socket */ - if ((ptr->hosts[server_key].fd= socket(AF_INET, SOCK_DGRAM, 0)) < 0) - { - ptr->cached_errno= errno; - return MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE; } } - return MEMCACHED_SUCCESS; } -static memcached_return tcp_connect(memcached_st *ptr, unsigned int server_key) +static memcached_return network_connect(memcached_server_st *ptr) { - if (ptr->hosts[server_key].fd == -1) + if (ptr->fd == -1) { struct addrinfo *use; /* Old connection junk still is in the structure */ - WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0); + WATCHPOINT_ASSERT(ptr->cursor_active == 0); - if (ptr->hosts[server_key].sockaddr_inited == MEMCACHED_NOT_ALLOCATED || - (!(ptr->flags & MEM_USE_CACHE_LOOKUPS))) + if (ptr->sockaddr_inited == MEMCACHED_NOT_ALLOCATED || + (!(ptr->root->flags & MEM_USE_CACHE_LOOKUPS))) { memcached_return rc; - rc= set_hostinfo(&ptr->hosts[server_key]); + rc= set_hostinfo(ptr); if (rc != MEMCACHED_SUCCESS) return rc; - ptr->hosts[server_key].sockaddr_inited= MEMCACHED_ALLOCATED; + ptr->sockaddr_inited= MEMCACHED_ALLOCATED; } - use= ptr->hosts[server_key].address_info; + use= ptr->address_info; /* Create the socket */ - if ((ptr->hosts[server_key].fd= socket(use->ai_family, - use->ai_socktype, - use->ai_protocol)) < 0) - { - ptr->cached_errno= errno; - WATCHPOINT_ERRNO(errno); - return MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE; - } - - if (ptr->flags & MEM_NO_BLOCK) - { - int error; - struct linger linger; - struct timeval waittime; - - waittime.tv_sec= 10; - waittime.tv_usec= 0; - - linger.l_onoff= 1; - linger.l_linger= MEMCACHED_DEFAULT_TIMEOUT; - error= setsockopt(ptr->hosts[server_key].fd, SOL_SOCKET, SO_LINGER, - &linger, (socklen_t)sizeof(struct linger)); - WATCHPOINT_ASSERT(error == 0); - - error= setsockopt(ptr->hosts[server_key].fd, SOL_SOCKET, SO_SNDTIMEO, - &waittime, (socklen_t)sizeof(struct timeval)); - WATCHPOINT_ASSERT(error == 0); - - error= setsockopt(ptr->hosts[server_key].fd, SOL_SOCKET, SO_RCVTIMEO, - &waittime, (socklen_t)sizeof(struct timeval)); - WATCHPOINT_ASSERT(error == 0); - } - - if (ptr->flags & MEM_TCP_NODELAY) + while (use != NULL) { - int flag= 1; - int error; - - error= setsockopt(ptr->hosts[server_key].fd, IPPROTO_TCP, TCP_NODELAY, - &flag, (socklen_t)sizeof(int)); - WATCHPOINT_ASSERT(error == 0); - } - - if (ptr->send_size) - { - int error; - - error= setsockopt(ptr->hosts[server_key].fd, SOL_SOCKET, SO_SNDBUF, - &ptr->send_size, (socklen_t)sizeof(int)); - WATCHPOINT_ASSERT(error == 0); - } - - if (ptr->recv_size) - { - int error; - - error= setsockopt(ptr->hosts[server_key].fd, SOL_SOCKET, SO_SNDBUF, - &ptr->recv_size, (socklen_t)sizeof(int)); - WATCHPOINT_ASSERT(error == 0); - } - - /* For the moment, not getting a nonblocking mode will not be fatal */ - if (ptr->flags & MEM_NO_BLOCK) - { - int flags; - - flags= fcntl(ptr->hosts[server_key].fd, F_GETFL, 0); - if (flags != -1) + if ((ptr->fd= socket(use->ai_family, + use->ai_socktype, + use->ai_protocol)) < 0) { - (void)fcntl(ptr->hosts[server_key].fd, F_SETFL, flags | O_NONBLOCK); + ptr->cached_errno= errno; + WATCHPOINT_ERRNO(errno); + return MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE; } - } + (void)set_socket_options(ptr); - /* connect to server */ + /* connect to server */ test_connect: - if (connect(ptr->hosts[server_key].fd, - use->ai_addr, - use->ai_addrlen) < 0) - { - switch (errno) { - /* We are spinning waiting on connect */ - case EALREADY: - case EINPROGRESS: - case EINTR: - goto test_connect; - case EISCONN: /* We were spinning waiting on connect */ - break; - default: - ptr->cached_errno= errno; - WATCHPOINT_ERRNO(ptr->cached_errno); - return MEMCACHED_ERRNO; + if (connect(ptr->fd, + use->ai_addr, + use->ai_addrlen) < 0) + { + switch (errno) { + /* We are spinning waiting on connect */ + case EALREADY: + case EINPROGRESS: + case EINTR: + goto test_connect; + case EISCONN: /* We were spinning waiting on connect */ + break; + default: + ptr->cached_errno= errno; + WATCHPOINT_ERRNO(ptr->cached_errno); + close(ptr->fd); + ptr->fd= -1; + } } - ptr->connected++; + else + { + WATCHPOINT_ASSERT(ptr->cursor_active == 0); + return MEMCACHED_SUCCESS; + } + use = use->ai_next; } - - WATCHPOINT_ASSERT(ptr->hosts[server_key].cursor_active == 0); } - return MEMCACHED_SUCCESS; + if (ptr->fd == -1) + return MEMCACHED_ERRNO; /* The last error should be from connect() */ + + return MEMCACHED_SUCCESS; /* The last error should be from connect() */ } -memcached_return memcached_connect(memcached_st *ptr, unsigned int server_key) +memcached_return memcached_connect(memcached_server_st *ptr) { memcached_return rc= MEMCACHED_NO_SERVERS; LIBMEMCACHED_MEMCACHED_CONNECT_START(); - if (ptr->connected == ptr->number_of_hosts && ptr->number_of_hosts) - return MEMCACHED_SUCCESS; - - if (ptr->hosts == NULL || ptr->number_of_hosts == 0) - return MEMCACHED_NO_SERVERS; - /* We need to clean up the multi startup piece */ - switch (ptr->hosts[server_key].type) + switch (ptr->type) { case MEMCACHED_CONNECTION_UNKNOWN: WATCHPOINT_ASSERT(0); rc= MEMCACHED_NOT_SUPPORTED; break; case MEMCACHED_CONNECTION_UDP: - rc= udp_connect(ptr, server_key); - break; case MEMCACHED_CONNECTION_TCP: - rc= tcp_connect(ptr, server_key); + rc= network_connect(ptr); break; case MEMCACHED_CONNECTION_UNIX_SOCKET: - rc= unix_socket_connect(ptr, server_key); + rc= unix_socket_connect(ptr); break; + default: + WATCHPOINT_ASSERT(0); } if (rc != MEMCACHED_SUCCESS)