#include <libmemcached/common.h>
+
#include <cassert>
#include <ctime>
#include <sys/time.h>
-static memcached_return_t connect_poll(memcached_server_st *ptr)
+#ifndef SOCK_CLOEXEC
+# define SOCK_CLOEXEC 0
+#endif
+
+#ifndef SOCK_NONBLOCK
+# define SOCK_NONBLOCK 0
+#endif
+
+#ifndef FD_CLOEXEC
+# define FD_CLOEXEC 0
+#endif
+
+static memcached_return_t connect_poll(org::libmemcached::Instance* server)
{
struct pollfd fds[1];
- fds[0].fd = ptr->fd;
- fds[0].events = POLLOUT;
+ fds[0].fd= server->fd;
+ fds[0].events= POLLOUT;
- int error;
size_t loop_max= 5;
- while (--loop_max) // Should only loop on cases of ERESTART or EINTR
+ if (server->root->poll_timeout == 0)
{
- error= poll(fds, 1, ptr->root->connect_timeout);
+ return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT);
+ }
- switch (error)
+ while (--loop_max) // Should only loop on cases of ERESTART or EINTR
+ {
+ int number_of;
+ if ((number_of= poll(fds, 1, server->root->connect_timeout)) <= 0)
{
- case 1:
+ if (number_of == -1)
{
- int err;
- socklen_t len= sizeof (err);
- (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len);
+ int local_errno= get_socket_errno(); // We cache in case closesocket() modifies errno
+ switch (local_errno)
+ {
+#ifdef TARGET_OS_LINUX
+ case ERESTART:
+#endif
+ case EINTR:
+ continue;
+
+ case EFAULT:
+ case ENOMEM:
+ return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
+
+ case EINVAL:
+ return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid"));
+
+ default: // This should not happen
+ if (fds[0].revents & POLLERR)
+ {
+ int err;
+ 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;
+ }
+ local_errno= err;
+ }
+ }
+
+ 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, local_errno, MEMCACHED_AT);
+ }
+ }
+ assert(number_of == 0);
+
+ server->io_wait_count.timeouts++;
+ return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT);
+ }
+ if (fds[0].revents & POLLERR or
+ fds[0].revents & POLLHUP or
+ fds[0].revents & POLLNVAL)
+ {
+ int err;
+ socklen_t len= sizeof (err);
+ if (getsockopt(fds[0].fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0)
+ {
// We check the value to see what happened wth the socket.
if (err == 0)
{
return MEMCACHED_SUCCESS;
}
- else
- {
- ptr->cached_errno= errno;
-
- return MEMCACHED_ERRNO;
- }
+ errno= err;
}
- case 0:
- return MEMCACHED_TIMEOUT;
- default: // A real error occurred and we need to completely bail
- WATCHPOINT_ERRNO(get_socket_errno());
- switch (get_socket_errno())
- {
-#ifdef TARGET_OS_LINUX
- case ERESTART:
-#endif
- case EINTR:
- continue;
- default:
- if (fds[0].revents & POLLERR)
- {
- int err;
- socklen_t len= sizeof (err);
- (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len);
- ptr->cached_errno= (err == 0) ? get_socket_errno() : err;
- }
- else
- {
- ptr->cached_errno= get_socket_errno();
- }
-
- (void)closesocket(ptr->fd);
- ptr->fd= INVALID_SOCKET;
- return MEMCACHED_ERRNO;
- }
+ return memcached_set_errno(*server, err, MEMCACHED_AT);
}
+ assert(fds[0].revents & POLLIN or fds[0].revents & POLLOUT);
+
+ return MEMCACHED_SUCCESS;
}
// This should only be possible from ERESTART or EINTR;
- ptr->cached_errno= get_socket_errno();
-
- return MEMCACHED_ERRNO;
+ return memcached_set_errno(*server, get_socket_errno(), MEMCACHED_AT);
}
-static memcached_return_t set_hostinfo(memcached_server_st *server)
+static memcached_return_t set_hostinfo(org::libmemcached::Instance* server)
{
- char str_port[NI_MAXSERV];
-
- assert(! server->address_info); // We cover the case where a programming mistake has been made.
+ assert(server->type != MEMCACHED_CONNECTION_UNIX_SOCKET);
if (server->address_info)
{
freeaddrinfo(server->address_info);
server->address_info_next= NULL;
}
- int length= snprintf(str_port, NI_MAXSERV, "%u", (uint32_t)server->port);
- if (length >= NI_MAXSERV || length < 0)
+ char str_port[NI_MAXSERV];
+ int length= snprintf(str_port, NI_MAXSERV, "%u", uint32_t(server->port()));
+ if (length >= NI_MAXSERV or length <= 0)
+ {
return MEMCACHED_FAILURE;
+ }
struct addrinfo hints;
memset(&hints, 0, sizeof(struct addrinfo));
#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;
hints.ai_protocol= IPPROTO_TCP;
}
+ assert(server->address_info == NULL);
+ assert(server->address_info_next == NULL);
int errcode;
switch(errcode= getaddrinfo(server->hostname, str_port, &hints, &server->address_info))
{
break;
case EAI_AGAIN:
+ if (server->address_info)
+ {
+ freeaddrinfo(server->address_info);
+ server->address_info= NULL;
+ server->address_info_next= NULL;
+ }
return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
case EAI_SYSTEM:
- return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_SYSTEM)"));
+ if (server->address_info)
+ {
+ freeaddrinfo(server->address_info);
+ server->address_info= NULL;
+ server->address_info_next= NULL;
+ }
+ return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_SYSTEM)"));
case EAI_BADFLAGS:
+ if (server->address_info)
+ {
+ freeaddrinfo(server->address_info);
+ server->address_info= NULL;
+ server->address_info_next= NULL;
+ }
return memcached_set_error(*server, MEMCACHED_INVALID_ARGUMENTS, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_BADFLAGS)"));
case EAI_MEMORY:
- return memcached_set_error(*server, MEMCACHED_ERRNO, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
+ if (server->address_info)
+ {
+ freeaddrinfo(server->address_info);
+ server->address_info= NULL;
+ server->address_info_next= NULL;
+ }
+ return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
default:
{
- WATCHPOINT_STRING(server->hostname);
- WATCHPOINT_STRING(gai_strerror(e));
+ if (server->address_info)
+ {
+ freeaddrinfo(server->address_info);
+ server->address_info= NULL;
+ server->address_info_next= NULL;
+ }
return memcached_set_error(*server, MEMCACHED_HOST_LOOKUP_FAILURE, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
}
}
server->address_info_next= server->address_info;
+ server->state= MEMCACHED_SERVER_STATE_ADDRINFO;
return MEMCACHED_SUCCESS;
}
-static inline memcached_return_t set_socket_nonblocking(memcached_server_st *ptr)
+static inline void set_socket_nonblocking(org::libmemcached::Instance* server)
{
#ifdef WIN32
- u_long arg = 1;
- if (ioctlsocket(ptr->fd, FIONBIO, &arg) == SOCKET_ERROR)
+ u_long arg= 1;
+ if (ioctlsocket(server->fd, FIONBIO, &arg) == SOCKET_ERROR)
{
- return memcached_set_errno(*ptr, get_socket_errno(), NULL);
+ memcached_set_errno(*server, get_socket_errno(), NULL);
}
#else
int flags;
- do
+ if (SOCK_NONBLOCK == 0)
{
- flags= fcntl(ptr->fd, F_GETFL, 0);
- }
- while (flags == -1 && (errno == EINTR || errno == EAGAIN));
-
- unlikely (flags == -1)
- {
- return memcached_set_errno(*ptr, errno, NULL);
- }
- else if ((flags & O_NONBLOCK) == 0)
- {
- int rval;
-
do
{
- rval= fcntl(ptr->fd, F_SETFL, flags | O_NONBLOCK);
- }
- while (rval == -1 && (errno == EINTR || errno == EAGAIN));
+ flags= fcntl(server->fd, F_GETFL, 0);
+ } while (flags == -1 && (errno == EINTR || errno == EAGAIN));
- unlikely (rval == -1)
+ if (flags == -1)
+ {
+ memcached_set_errno(*server, errno, NULL);
+ }
+ else if ((flags & O_NONBLOCK) == 0)
{
- return memcached_set_errno(*ptr, errno, NULL);
+ int rval;
+
+ do
+ {
+ rval= fcntl(server->fd, F_SETFL, flags | O_NONBLOCK);
+ } while (rval == -1 && (errno == EINTR or errno == EAGAIN));
+
+ if (rval == -1)
+ {
+ memcached_set_errno(*server, errno, NULL);
+ }
}
}
#endif
- return MEMCACHED_SUCCESS;
}
-static memcached_return_t set_socket_options(memcached_server_st *ptr)
+static void set_socket_options(org::libmemcached::Instance* server)
{
- WATCHPOINT_ASSERT(ptr->fd != -1);
+ assert_msg(server->fd != INVALID_SOCKET, "invalid socket was passed to set_socket_options()");
- if (ptr->type == MEMCACHED_CONNECTION_UDP)
- return MEMCACHED_SUCCESS;
+ if (memcached_is_udp(server->root))
+ {
+ return;
+ }
#ifdef HAVE_SNDTIMEO
- if (ptr->root->snd_timeout)
+ if (server->root->snd_timeout > 0)
{
- int error;
struct timeval waittime;
- waittime.tv_sec= 0;
- waittime.tv_usec= ptr->root->snd_timeout;
+ waittime.tv_sec= server->root->snd_timeout / 1000000;
+ waittime.tv_usec= server->root->snd_timeout % 1000000;
- error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDTIMEO,
- &waittime, (socklen_t)sizeof(struct timeval));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_SNDTIMEO,
+ &waittime, (socklen_t)sizeof(struct timeval));
+ (void)error;
+ assert(error == 0);
}
#endif
#ifdef HAVE_RCVTIMEO
- if (ptr->root->rcv_timeout)
+ if (server->root->rcv_timeout > 0)
{
- int error;
struct timeval waittime;
- waittime.tv_sec= 0;
- waittime.tv_usec= ptr->root->rcv_timeout;
+ waittime.tv_sec= server->root->rcv_timeout / 1000000;
+ waittime.tv_usec= server->root->rcv_timeout % 1000000;
- error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVTIMEO,
- &waittime, (socklen_t)sizeof(struct timeval));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_RCVTIMEO,
+ &waittime, (socklen_t)sizeof(struct timeval));
+ (void)(error);
+ assert(error == 0);
}
#endif
#if defined(__MACH__) && defined(__APPLE__) || defined(__FreeBSD__)
{
- int set = 1;
- int error= setsockopt(ptr->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
+ int set= 1;
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
+
+ assert(error == 0);
// This is not considered a fatal error
if (error == -1)
{
- WATCHPOINT_ERRNO(get_socket_errno());
+#if 0
perror("setsockopt(SO_NOSIGPIPE)");
+#endif
}
}
#endif
- if (ptr->root->flags.no_block)
+ if (server->root->flags.no_block)
{
- int error;
struct linger linger;
linger.l_onoff= 1;
linger.l_linger= 0; /* By default on close() just drop the socket */
- error= setsockopt(ptr->fd, SOL_SOCKET, SO_LINGER,
- &linger, (socklen_t)sizeof(struct linger));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_LINGER,
+ &linger, (socklen_t)sizeof(struct linger));
+ (void)(error);
+ assert(error == 0);
}
- if (ptr->root->flags.tcp_nodelay)
+ if (server->root->flags.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 (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, IPPROTO_TCP, TCP_NODELAY,
+ &flag, (socklen_t)sizeof(int));
+ (void)(error);
+ assert(error == 0);
}
- if (ptr->root->flags.tcp_keepalive)
+ if (server->root->flags.tcp_keepalive)
{
int flag= 1;
- int error;
- error= setsockopt(ptr->fd, SOL_SOCKET, SO_KEEPALIVE,
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_KEEPALIVE,
&flag, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ (void)(error);
+ assert(error == 0);
}
#ifdef TCP_KEEPIDLE
- if (ptr->root->tcp_keepidle > 0)
+ if (server->root->tcp_keepidle > 0)
{
- int error;
-
- error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_KEEPIDLE,
- &ptr->root->tcp_keepidle, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, IPPROTO_TCP, TCP_KEEPIDLE,
+ &server->root->tcp_keepidle, (socklen_t)sizeof(int));
+ (void)(error);
+ assert(error == 0);
}
#endif
- if (ptr->root->send_size > 0)
+ if (server->root->send_size > 0)
{
- int error;
-
- error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF,
- &ptr->root->send_size, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_SNDBUF,
+ &server->root->send_size, (socklen_t)sizeof(int));
+ (void)(error);
+ assert(error == 0);
}
- if (ptr->root->recv_size > 0)
+ if (server->root->recv_size > 0)
{
- int error;
-
- error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVBUF,
- &ptr->root->recv_size, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
- if (error)
- return MEMCACHED_FAILURE;
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_RCVBUF,
+ &server->root->recv_size, (socklen_t)sizeof(int));
+ (void)(error);
+ assert(error == 0);
}
-
/* libmemcached will always use nonblocking IO to avoid write deadlocks */
- return set_socket_nonblocking(ptr);
+ set_socket_nonblocking(server);
}
-static memcached_return_t unix_socket_connect(memcached_server_st *ptr)
+static memcached_return_t unix_socket_connect(org::libmemcached::Instance* server)
{
#ifndef WIN32
- WATCHPOINT_ASSERT(ptr->fd == -1);
+ WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
- if ((ptr->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
+ int type= SOCK_STREAM;
+ if (SOCK_CLOEXEC)
{
- return memcached_set_errno(*ptr, errno, NULL);
+ type|= SOCK_CLOEXEC;
+ }
+
+ if (SOCK_NONBLOCK)
+ {
+ type|= SOCK_NONBLOCK;
+ }
+
+ if ((server->fd= socket(AF_UNIX, type, 0)) < 0)
+ {
+ memcached_set_errno(*server, errno, NULL);
+ return MEMCACHED_CONNECTION_FAILURE;
}
struct sockaddr_un servAddr;
memset(&servAddr, 0, sizeof (struct sockaddr_un));
servAddr.sun_family= AF_UNIX;
- strncpy(servAddr.sun_path, ptr->hostname, sizeof(servAddr.sun_path)); /* Copy filename */
+ strncpy(servAddr.sun_path, server->hostname, sizeof(servAddr.sun_path)); /* Copy filename */
-test_connect:
- if (connect(ptr->fd,
- (struct sockaddr *)&servAddr,
- sizeof(servAddr)) < 0)
- {
- switch (errno)
+ do {
+ if (connect(server->fd, (struct sockaddr *)&servAddr, sizeof(servAddr)) < 0)
{
- case EINPROGRESS:
- case EALREADY:
- case EINTR:
- goto test_connect;
- case EISCONN: /* We were spinning waiting on connect */
- break;
- default:
- WATCHPOINT_ERRNO(errno);
- ptr->cached_errno= errno;
- return MEMCACHED_ERRNO;
+ switch (errno)
+ {
+ case EINPROGRESS:
+ case EALREADY:
+ case EINTR:
+ continue;
+
+ case EISCONN: /* We were spinning waiting on connect */
+ {
+ assert(0); // Programmer error
+ break;
+ }
+
+ default:
+ WATCHPOINT_ERRNO(errno);
+ memcached_set_errno(*server, errno, MEMCACHED_AT);
+ return MEMCACHED_CONNECTION_FAILURE;
+ }
}
- }
+ } while (0);
+ server->state= MEMCACHED_SERVER_STATE_CONNECTED;
- WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
+ WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
return MEMCACHED_SUCCESS;
#else
- (void)ptr;
+ (void)server;
return MEMCACHED_NOT_SUPPORTED;
#endif
}
-static memcached_return_t network_connect(memcached_server_st *ptr)
+static memcached_return_t network_connect(org::libmemcached::Instance* server)
{
bool timeout_error_occured= false;
- WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
- WATCHPOINT_ASSERT(ptr->cursor_active == 0);
+ WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
+ WATCHPOINT_ASSERT(server->cursor_active_ == 0);
- if (not ptr->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)
{
- memcached_return_t rc;
- uint32_t counter= 5;
- while (--counter)
- {
- if ((rc= set_hostinfo(ptr)) != MEMCACHED_TIMEOUT)
- break;
-
-#ifndef WIN32
- struct timespec dream, rem;
-
- dream.tv_nsec= 1000;
- dream.tv_sec= 0;
-
- nanosleep(&dream, &rem);
-#endif
- }
+ WATCHPOINT_ASSERT(server->state == MEMCACHED_SERVER_STATE_NEW);
+ server->address_info_next= NULL;
+ memcached_return_t rc= set_hostinfo(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 (ptr->address_info_next && ptr->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 (ptr->type == MEMCACHED_CONNECTION_UDP && ptr->address_info_next->ai_family != AF_INET)
+ if (memcached_is_udp(server->root) and server->address_info_next->ai_family != AF_INET)
{
- ptr->address_info_next= ptr->address_info_next->ai_next;
+ server->address_info_next= server->address_info_next->ai_next;
continue;
}
- if ((ptr->fd= socket(ptr->address_info_next->ai_family,
- ptr->address_info_next->ai_socktype,
- ptr->address_info_next->ai_protocol)) < 0)
+ int type= server->address_info_next->ai_socktype;
+ if (SOCK_CLOEXEC)
+ {
+ type|= SOCK_CLOEXEC;
+ }
+
+ if (SOCK_NONBLOCK)
{
- return memcached_set_errno(*ptr, get_socket_errno(), NULL);
+ type|= SOCK_NONBLOCK;
}
- (void)set_socket_options(ptr);
+ if ((server->fd= socket(server->address_info_next->ai_family,
+ type,
+ server->address_info_next->ai_protocol)) < 0)
+ {
+ return memcached_set_errno(*server, get_socket_errno(), NULL);
+ }
+
+ // If SOCK_CLOEXEC exists then we don't need to call the following
+ if (SOCK_CLOEXEC == 0)
+ {
+ if (FD_CLOEXEC)
+ {
+ int rval;
+ do
+ {
+ rval= fcntl (server->fd, F_SETFD, FD_CLOEXEC);
+ } while (rval == -1 && (errno == EINTR or errno == EAGAIN));
+ }
+ }
+
+ set_socket_options(server);
/* connect to server */
- if ((connect(ptr->fd, ptr->address_info_next->ai_addr, ptr->address_info_next->ai_addrlen) != SOCKET_ERROR))
+ if ((connect(server->fd, server->address_info_next->ai_addr, server->address_info_next->ai_addrlen) != SOCKET_ERROR))
{
- break; // Success
+ server->state= MEMCACHED_SERVER_STATE_CONNECTED;
+ return MEMCACHED_SUCCESS;
}
/* An error occurred */
- ptr->cached_errno= get_socket_errno();
- switch (ptr->cached_errno)
+ switch (get_socket_errno())
{
+ case ETIMEDOUT:
+ 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
{
- memcached_return_t rc;
- rc= connect_poll(ptr);
+ server->state= MEMCACHED_SERVER_STATE_IN_PROGRESS;
+ memcached_return_t rc= connect_poll(server);
+ if (memcached_success(rc))
+ {
+ server->state= MEMCACHED_SERVER_STATE_CONNECTED;
+ return MEMCACHED_SUCCESS;
+ }
+
+ // A timeout here is treated as an error, we will not retry
if (rc == MEMCACHED_TIMEOUT)
+ {
timeout_error_occured= true;
-
- if (rc == MEMCACHED_SUCCESS)
- break;
+ }
}
+ break;
case EISCONN: // we are connected :-)
+ WATCHPOINT_ASSERT(0); // This is a programmer's error
break;
case EINTR: // Special case, we retry ai_addr
- (void)closesocket(ptr->fd);
- ptr->fd= INVALID_SOCKET;
+ WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
+ (void)closesocket(server->fd);
+ server->fd= INVALID_SOCKET;
continue;
default:
- (void)closesocket(ptr->fd);
- ptr->fd= INVALID_SOCKET;
- ptr->address_info_next= ptr->address_info_next->ai_next;
break;
}
+
+ WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
+ (void)closesocket(server->fd);
+ server->fd= INVALID_SOCKET;
+ server->address_info_next= server->address_info_next->ai_next;
}
- if (ptr->fd == INVALID_SOCKET)
- {
- WATCHPOINT_STRING("Never got a good file descriptor");
+ WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
- /* Failed to connect. schedule next retry */
- if (ptr->root->retry_timeout)
+ if (timeout_error_occured)
+ {
+ if (server->fd != INVALID_SOCKET)
{
- struct timeval next_time;
-
- if (gettimeofday(&next_time, NULL) == 0)
- ptr->next_retry= next_time.tv_sec + ptr->root->retry_timeout;
+ (void)closesocket(server->fd);
+ server->fd= INVALID_SOCKET;
}
+ }
- if (timeout_error_occured)
- return MEMCACHED_TIMEOUT;
+ WATCHPOINT_STRING("Never got a good file descriptor");
- return MEMCACHED_ERRNO; /* The last error should be from connect() */
+ if (memcached_has_current_error(*server))
+ {
+ return memcached_instance_error_return(server);
+ }
+
+ if (timeout_error_occured and server->state < MEMCACHED_SERVER_STATE_IN_PROGRESS)
+ {
+ return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT);
}
- return MEMCACHED_SUCCESS; /* The last error should be from connect() */
+ 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;
-#if 0
- WATCHPOINT_STRING(self->hostname);
- WATCHPOINT_NUMBER(self->port);
- WATCHPOINT_ERRNO(self->cached_errno);
-#endif
- 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 ptr)
+ 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(org::libmemcached::Instance* server, bool& in_timeout)
{
- memcached_return_t rc= MEMCACHED_NO_SERVERS;
+ struct timeval curr_time;
+ bool _gettime_success= (gettimeofday(&curr_time, NULL) == 0);
- if (ptr->fd != INVALID_SOCKET)
- return MEMCACHED_SUCCESS;
+ /*
+ If we hit server_failure_limit then something is completely wrong about the server.
- LIBMEMCACHED_MEMCACHED_CONNECT_START();
-
- /* both retry_timeout and server_failure_limit must be set in order to delay retrying a server on error. */
- WATCHPOINT_ASSERT(ptr->root);
- if (ptr->root->retry_timeout && ptr->next_retry)
+ 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)
{
- struct timeval curr_time;
+ /*
+ We just auto_eject if we hit this point
+ */
+ if (_is_auto_eject_host(server->root))
+ {
+ set_last_disconnected_host(server);
- gettimeofday(&curr_time, NULL);
+ // 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;
- // We should optimize this to remove the allocation if the server was
- // the last server to die
- if (ptr->next_retry > curr_time.tv_sec)
- {
- set_last_disconnected_host(ptr);
+ // We only retry dead servers once before assuming failure again
+ server->server_failure_counter= server->root->server_failure_limit -1;
+ }
+
+ 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;
- return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
+ // Sanity check/setting
+ if (server->next_retry == 0)
+ {
+ server->next_retry= 1;
}
}
- // If we are over the counter failure, we just fail. Reject host only
- // works if you have a set number of failures.
- if (ptr->root->server_failure_limit && ptr->server_failure_counter >= ptr->root->server_failure_limit)
+ if (server->state == MEMCACHED_SERVER_STATE_IN_TIMEOUT)
{
- set_last_disconnected_host(ptr);
-
- // @todo fix this by fixing behavior to no longer make use of
- // memcached_st
- if (_is_auto_eject_host(ptr->root))
+ /*
+ 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)
{
- run_distribution((memcached_st *)ptr->root);
+ server->state= MEMCACHED_SERVER_STATE_NEW;
+ }
+ else
+ {
+ return memcached_set_error(*server, MEMCACHED_SERVER_TEMPORARILY_DISABLED, MEMCACHED_AT);
}
- return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
+ in_timeout= true;
}
- /* We need to clean up the multi startup piece */
- switch (ptr->type)
+ return MEMCACHED_SUCCESS;
+}
+
+static memcached_return_t _memcached_connect(org::libmemcached::Instance* server, const bool set_last_disconnected)
+{
+ assert(server);
+ if (server->fd != INVALID_SOCKET)
{
- case MEMCACHED_CONNECTION_UNKNOWN:
- WATCHPOINT_ASSERT(0);
- rc= MEMCACHED_NOT_SUPPORTED;
- break;
+ return MEMCACHED_SUCCESS;
+ }
+ 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;
+ }
+
+ 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"));
+ }
+
+ if (server->hostname[0] == '/')
+ {
+ server->type= MEMCACHED_CONNECTION_UNIX_SOCKET;
+ }
+
+ /* We need to clean up the multi startup piece */
+ switch (server->type)
+ {
case MEMCACHED_CONNECTION_UDP:
case MEMCACHED_CONNECTION_TCP:
- rc= network_connect(ptr);
-#ifdef LIBMEMCACHED_WITH_SASL_SUPPORT
- if (ptr->fd != INVALID_SOCKET && ptr->root->sasl.callbacks)
+ rc= network_connect(server);
+
+ if (LIBMEMCACHED_WITH_SASL_SUPPORT)
{
- rc= memcached_sasl_authenticate_connection(ptr);
- if (memcached_failed(rc))
+ if (server->fd != INVALID_SOCKET and server->root->sasl.callbacks)
{
- (void)closesocket(ptr->fd);
- ptr->fd= INVALID_SOCKET;
+ rc= memcached_sasl_authenticate_connection(server);
+ if (memcached_failed(rc) and server->fd != INVALID_SOCKET)
+ {
+ WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
+ (void)closesocket(server->fd);
+ server->fd= INVALID_SOCKET;
+ }
}
}
-#endif
break;
case MEMCACHED_CONNECTION_UNIX_SOCKET:
- rc= unix_socket_connect(ptr);
+ rc= unix_socket_connect(server);
break;
-
- case MEMCACHED_CONNECTION_MAX:
- default:
- WATCHPOINT_ASSERT(0);
}
if (memcached_success(rc))
{
- ptr->server_failure_counter= 0;
- ptr->next_retry= 0;
+ server->mark_server_as_clean();
+ memcached_version_instance(server);
+ return rc;
}
- else
+ else if (set_last_disconnected)
{
- memcached_set_error(*ptr, rc, MEMCACHED_AT);
- ptr->server_failure_counter++;
- set_last_disconnected_host(ptr);
- }
+ set_last_disconnected_host(server);
+ if (memcached_has_current_error(*server))
+ {
+ memcached_mark_server_for_timeout(server);
+ assert(memcached_failed(memcached_instance_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(org::libmemcached::Instance* server)
+{
+ if (server and server->root and server->root->state.is_parsing)
+ {
+ return MEMCACHED_SUCCESS;
+ }
+
+ return _memcached_connect(server, false);
+}
+
+memcached_return_t memcached_connect(org::libmemcached::Instance* server)
+{
+ return _memcached_connect(server, true);
+}
projects / awesomized / libmemcached / blobdiff