X-Git-Url: https://git.m6w6.name/?a=blobdiff_plain;f=libmemcached%2Fconnect.cc;h=9d50e5c0dc05ecc1ad1f0ecd754077a1ff9eda15;hb=ad7ce5a1c9769432d8105c1a3a3a67b12d8bd731;hp=f90cb6b057958ec3db6feedd35cbbf9c1914a7af;hpb=0fe409a0db64fc9d24a44ac363c502ea9e98857e;p=m6w6%2Flibmemcached diff --git a/libmemcached/connect.cc b/libmemcached/connect.cc index f90cb6b0..9d50e5c0 100644 --- a/libmemcached/connect.cc +++ b/libmemcached/connect.cc @@ -37,6 +37,8 @@ #include + +#include #include #include @@ -62,18 +64,21 @@ static memcached_return_t connect_poll(memcached_server_st *server) { int err; socklen_t len= sizeof (err); - (void)getsockopt(server->fd, SOL_SOCKET, SO_ERROR, &err, &len); - - // We check the value to see what happened wth the socket. - if (err == 0) + if (getsockopt(server->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0) { - return MEMCACHED_SUCCESS; + // We check the value to see what happened wth the socket. + if (err == 0) + { + return MEMCACHED_SUCCESS; + } + errno= err; } return memcached_set_errno(*server, err, MEMCACHED_AT); } case 0: { + server->io_wait_count.timeouts++; return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT); } @@ -97,21 +102,26 @@ static memcached_return_t connect_poll(memcached_server_st *server) if (fds[0].revents & POLLERR) { int err; - socklen_t len= sizeof (err); - (void)getsockopt(server->fd, SOL_SOCKET, SO_ERROR, &err, &len); - memcached_set_errno(*server, (err == 0) ? get_socket_errno() : err, MEMCACHED_AT); - } - else - { - memcached_set_errno(*server, get_socket_errno(), MEMCACHED_AT); + socklen_t len= sizeof(err); + if (getsockopt(server->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0) + { + if (err == 0) + { + // This should never happen, if it does? Punt. + continue; + } + errno= err; + } } + int local_errno= get_socket_errno(); // We cache in case closesocket() modifies errno + assert_msg(server->fd != INVALID_SOCKET, "poll() was passed an invalid file descriptor"); (void)closesocket(server->fd); server->fd= INVALID_SOCKET; server->state= MEMCACHED_SERVER_STATE_NEW; - return memcached_set_errno(*server, get_socket_errno(), MEMCACHED_AT); + return memcached_set_errno(*server, local_errno, MEMCACHED_AT); } } } @@ -122,6 +132,7 @@ static memcached_return_t connect_poll(memcached_server_st *server) static memcached_return_t set_hostinfo(memcached_server_st *server) { + assert(server->type != MEMCACHED_CONNECTION_UNIX_SOCKET); if (server->address_info) { freeaddrinfo(server->address_info); @@ -142,7 +153,7 @@ static memcached_return_t set_hostinfo(memcached_server_st *server) #if 0 hints.ai_family= AF_INET; #endif - if (server->type == MEMCACHED_CONNECTION_UDP) + if (memcached_is_udp(server->root)) { hints.ai_protocol= IPPROTO_UDP; hints.ai_socktype= SOCK_DGRAM; @@ -210,9 +221,9 @@ static inline void set_socket_nonblocking(memcached_server_st *server) do { rval= fcntl(server->fd, F_SETFL, flags | O_NONBLOCK); - } while (rval == -1 && (errno == EINTR || errno == EAGAIN)); + } while (rval == -1 && (errno == EINTR or errno == EAGAIN)); - unlikely (rval == -1) + if (rval == -1) { memcached_set_errno(*server, errno, NULL); } @@ -222,9 +233,9 @@ static inline void set_socket_nonblocking(memcached_server_st *server) static void set_socket_options(memcached_server_st *server) { - assert_msg(server->fd != -1, "invalid socket was passed to set_socket_options()"); + assert_msg(server->fd != INVALID_SOCKET, "invalid socket was passed to set_socket_options()"); - if (server->type == MEMCACHED_CONNECTION_UDP) + if (memcached_is_udp(server->root)) { return; } @@ -343,7 +354,7 @@ static void set_socket_options(memcached_server_st *server) static memcached_return_t unix_socket_connect(memcached_server_st *server) { #ifndef WIN32 - WATCHPOINT_ASSERT(server->fd == -1); + WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET); if ((server->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0) { @@ -398,9 +409,13 @@ static memcached_return_t network_connect(memcached_server_st *server) WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET); WATCHPOINT_ASSERT(server->cursor_active == 0); - if (not server->address_info) + /* + We want to check both of these because if address_info_next has been fully tried, we want to do a new lookup to make sure we have picked up on any new DNS information. + */ + if (server->address_info == NULL or server->address_info_next == NULL) { WATCHPOINT_ASSERT(server->state == MEMCACHED_SERVER_STATE_NEW); + server->address_info_next= NULL; memcached_return_t rc; uint32_t counter= 5; while (--counter) @@ -421,14 +436,22 @@ static memcached_return_t network_connect(memcached_server_st *server) } if (memcached_failed(rc)) + { return rc; + } + } + + if (server->address_info_next == NULL) + { + server->address_info_next= server->address_info; + server->state= MEMCACHED_SERVER_STATE_ADDRINFO; } /* Create the socket */ - while (server->address_info_next && server->fd == INVALID_SOCKET) + while (server->address_info_next and server->fd == INVALID_SOCKET) { /* Memcache server does not support IPV6 in udp mode, so skip if not ipv4 */ - if (server->type == MEMCACHED_CONNECTION_UDP && server->address_info_next->ai_family != AF_INET) + if (memcached_is_udp(server->root) and server->address_info_next->ai_family != AF_INET) { server->address_info_next= server->address_info_next->ai_next; continue; @@ -457,7 +480,10 @@ static memcached_return_t network_connect(memcached_server_st *server) timeout_error_occured= true; break; + case EAGAIN: +#if EWOULDBLOCK != EAGAIN case EWOULDBLOCK: +#endif case EINPROGRESS: // nonblocking mode - first return case EALREADY: // nonblocking mode - subsequent returns { @@ -510,16 +536,6 @@ static memcached_return_t network_connect(memcached_server_st *server) } WATCHPOINT_STRING("Never got a good file descriptor"); - /* Failed to connect. schedule next retry */ - if (server->root->retry_timeout) - { - struct timeval next_time; - - if (gettimeofday(&next_time, NULL) == 0) - { - server->next_retry= next_time.tv_sec + server->root->retry_timeout; - } - } if (memcached_has_current_error(*server)) { @@ -534,58 +550,105 @@ static memcached_return_t network_connect(memcached_server_st *server) return memcached_set_error(*server, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); /* The last error should be from connect() */ } -void set_last_disconnected_host(memcached_server_write_instance_st self) -{ - // const_cast - memcached_st *root= (memcached_st *)self->root; - memcached_server_free(root->last_disconnected_server); - root->last_disconnected_server= memcached_server_clone(NULL, self); -} +/* + backoff_handling() -memcached_return_t memcached_connect(memcached_server_write_instance_st server) + Based on time/failure count fail the connect without trying. This prevents waiting in a state where + we get caught spending cycles just waiting. +*/ +static memcached_return_t backoff_handling(memcached_server_write_instance_st server, bool& in_timeout) { - memcached_return_t rc= MEMCACHED_NO_SERVERS; + struct timeval curr_time; + bool _gettime_success= (gettimeofday(&curr_time, NULL) == 0); - if (server->fd != INVALID_SOCKET) + /* + If we hit server_failure_limit then something is completely wrong about the server. + + 1) If autoeject is enabled we do that. + 2) If not? We go into timeout again, there is much else to do :( + */ + if (server->server_failure_counter >= server->root->server_failure_limit) { - return MEMCACHED_SUCCESS; - } + /* + We just auto_eject if we hit this point + */ + if (_is_auto_eject_host(server->root)) + { + set_last_disconnected_host(server); - LIBMEMCACHED_MEMCACHED_CONNECT_START(); + // Retry dead servers if requested + if (_gettime_success and server->root->dead_timeout > 0) + { + server->next_retry= curr_time.tv_sec +server->root->dead_timeout; - /* both retry_timeout and server_failure_limit must be set in order to delay retrying a server on error. */ - WATCHPOINT_ASSERT(server->root); - if (server->root->retry_timeout and server->next_retry) - { - struct timeval curr_time; + // We only retry dead servers once before assuming failure again + server->server_failure_counter= server->root->server_failure_limit -1; + } - gettimeofday(&curr_time, NULL); + memcached_return_t rc; + if (memcached_failed(rc= run_distribution((memcached_st *)server->root))) + { + return memcached_set_error(*server, rc, MEMCACHED_AT, memcached_literal_param("Backoff handling failed during run_distribution")); + } + + return memcached_set_error(*server, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT); + } + + server->state= MEMCACHED_SERVER_STATE_IN_TIMEOUT; - // We should optimize this to remove the allocation if the server was - // the last server to die - if (server->next_retry > curr_time.tv_sec) + // Sanity check/setting + if (server->next_retry == 0) { - set_last_disconnected_host(server); + server->next_retry= 1; + } + } - return memcached_set_error(*server, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT); + if (server->state == MEMCACHED_SERVER_STATE_IN_TIMEOUT) + { + /* + If next_retry is less then our current time, then we reset and try everything again. + */ + if (_gettime_success and server->next_retry < curr_time.tv_sec) + { + server->state= MEMCACHED_SERVER_STATE_NEW; + } + else + { + return memcached_set_error(*server, MEMCACHED_SERVER_TEMPORARILY_DISABLED, MEMCACHED_AT); } + + in_timeout= true; + } + + return MEMCACHED_SUCCESS; +} + +static memcached_return_t _memcached_connect(memcached_server_write_instance_st server, const bool set_last_disconnected) +{ + if (server->fd != INVALID_SOCKET) + { + return MEMCACHED_SUCCESS; } - // If we are over the counter failure, we just fail. Reject host only - // works if you have a set number of failures. - if (server->root->server_failure_limit and server->server_failure_counter >= server->root->server_failure_limit) + LIBMEMCACHED_MEMCACHED_CONNECT_START(); + + bool in_timeout= false; + memcached_return_t rc; + if (memcached_failed(rc= backoff_handling(server, in_timeout))) { set_last_disconnected_host(server); + return rc; + } - // @todo fix this by fixing behavior to no longer make use of - // memcached_st - if (_is_auto_eject_host(server->root)) - { - run_distribution((memcached_st *)server->root); - } + if (LIBMEMCACHED_WITH_SASL_SUPPORT and server->root->sasl.callbacks and memcached_is_udp(server->root)) + { + return memcached_set_error(*server, MEMCACHED_INVALID_HOST_PROTOCOL, MEMCACHED_AT, memcached_literal_param("SASL is not supported for UDP connections")); + } - return memcached_set_error(*server, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT); + if (server->hostname[0] == '/') + { + server->type= MEMCACHED_CONNECTION_UNIX_SOCKET; } /* We need to clean up the multi startup piece */ @@ -594,6 +657,7 @@ memcached_return_t memcached_connect(memcached_server_write_instance_st server) case MEMCACHED_CONNECTION_UDP: case MEMCACHED_CONNECTION_TCP: rc= network_connect(server); + if (LIBMEMCACHED_WITH_SASL_SUPPORT) { if (server->fd != INVALID_SOCKET and server->root->sasl.callbacks) @@ -616,22 +680,42 @@ memcached_return_t memcached_connect(memcached_server_write_instance_st server) if (memcached_success(rc)) { - server->server_failure_counter= 0; - server->next_retry= 0; - } - else if (memcached_has_current_error(*server)) - { - server->server_failure_counter++; - set_last_disconnected_host(server); + memcached_mark_server_as_clean(server); + return rc; } - else + else if (set_last_disconnected) { - memcached_set_error(*server, rc, MEMCACHED_AT); - server->server_failure_counter++; set_last_disconnected_host(server); - } + if (memcached_has_current_error(*server)) + { + memcached_mark_server_for_timeout(server); + assert(memcached_failed(memcached_server_error_return(server))); + } + else + { + memcached_set_error(*server, rc, MEMCACHED_AT); + memcached_mark_server_for_timeout(server); + } - LIBMEMCACHED_MEMCACHED_CONNECT_END(); + LIBMEMCACHED_MEMCACHED_CONNECT_END(); + + if (in_timeout) + { + char buffer[1024]; + int snprintf_length= snprintf(buffer, sizeof(buffer), "%s:%d", server->hostname, int(server->port)); + return memcached_set_error(*server, MEMCACHED_SERVER_TEMPORARILY_DISABLED, MEMCACHED_AT, buffer, snprintf_length); + } + } return rc; } + +memcached_return_t memcached_connect_try(memcached_server_write_instance_st server) +{ + return _memcached_connect(server, false); +} + +memcached_return_t memcached_connect(memcached_server_write_instance_st server) +{ + return _memcached_connect(server, true); +}