#include <ctime>
#include <sys/time.h>
-static memcached_return_t connect_poll(memcached_server_st *server)
+#ifndef SOCK_CLOEXEC
+#define SOCK_CLOEXEC 0
+#endif
+
+static memcached_return_t connect_poll(memcached_instance_st *server)
{
struct pollfd fds[1];
fds[0].fd= server->fd;
while (--loop_max) // Should only loop on cases of ERESTART or EINTR
{
- int error= poll(fds, 1, server->root->connect_timeout);
- switch (error)
+ 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);
- if (getsockopt(server->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0)
+ int local_errno= get_socket_errno(); // We cache in case closesocket() modifies errno
+ switch (local_errno)
{
- // 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:
- {
- return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT);
- }
-
- default: // A real error occurred and we need to completely bail
- switch (get_socket_errno())
- {
#ifdef TARGET_OS_LINUX
- case ERESTART:
+ case ERESTART:
#endif
- case EINTR:
- continue;
+ case EINTR:
+ continue;
- case EFAULT:
- case ENOMEM:
- return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
+ 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"));
+ 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)
+ default: // This should not happen
+ if (fds[0].revents & POLLERR)
{
- if (err == 0)
+ int err;
+ socklen_t len= sizeof(err);
+ if (getsockopt(server->fd, SOL_SOCKET, SO_ERROR, &err, &len) == 0)
{
- // This should never happen, if it does? Punt.
- continue;
+ if (err == 0)
+ {
+ // This should never happen, if it does? Punt.
+ continue;
+ }
+ local_errno= err;
}
- 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, local_errno, MEMCACHED_AT);
+ }
+ }
+ assert(number_of == 0);
- 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;
+ server->io_wait_count.timeouts++;
+ return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT);
+ }
- return memcached_set_errno(*server, local_errno, 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;
+ }
+ errno= err;
}
+
+ 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;
return memcached_set_errno(*server, get_socket_errno(), MEMCACHED_AT);
}
-memcached_return_t set_hostinfo(memcached_server_st *server)
+static memcached_return_t set_hostinfo(memcached_instance_st *server)
{
+ assert(server->type != MEMCACHED_CONNECTION_UNIX_SOCKET);
if (server->address_info)
{
freeaddrinfo(server->address_info);
return MEMCACHED_SUCCESS;
}
-static inline void set_socket_nonblocking(memcached_server_st *server)
+static inline void set_socket_nonblocking(memcached_instance_st *server)
{
#ifdef WIN32
u_long arg= 1;
#endif
}
-static void set_socket_options(memcached_server_st *server)
+static void set_socket_options(memcached_instance_st *server)
{
assert_msg(server->fd != INVALID_SOCKET, "invalid socket was passed to set_socket_options()");
#ifdef HAVE_SNDTIMEO
if (server->root->snd_timeout)
{
- int error;
struct timeval waittime;
waittime.tv_sec= 0;
waittime.tv_usec= server->root->snd_timeout;
- error= setsockopt(server->fd, SOL_SOCKET, SO_SNDTIMEO,
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_SNDTIMEO,
&waittime, (socklen_t)sizeof(struct timeval));
- WATCHPOINT_ASSERT(error == 0);
+ assert(error == 0);
}
#endif
#ifdef HAVE_RCVTIMEO
if (server->root->rcv_timeout)
{
- int error;
struct timeval waittime;
waittime.tv_sec= 0;
waittime.tv_usec= server->root->rcv_timeout;
- error= setsockopt(server->fd, SOL_SOCKET, SO_RCVTIMEO,
- &waittime, (socklen_t)sizeof(struct timeval));
- WATCHPOINT_ASSERT(error == 0);
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_RCVTIMEO,
+ &waittime, (socklen_t)sizeof(struct timeval));
+ assert(error == 0);
}
#endif
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(server->fd, SOL_SOCKET, SO_LINGER,
- &linger, (socklen_t)sizeof(struct linger));
- WATCHPOINT_ASSERT(error == 0);
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_LINGER,
+ &linger, (socklen_t)sizeof(struct linger));
+ assert(error == 0);
}
if (server->root->flags.tcp_nodelay)
{
int flag= 1;
- int error;
- error= setsockopt(server->fd, IPPROTO_TCP, TCP_NODELAY,
- &flag, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
+ int error= setsockopt(server->fd, IPPROTO_TCP, TCP_NODELAY,
+ &flag, (socklen_t)sizeof(int));
+ assert(error == 0);
}
if (server->root->flags.tcp_keepalive)
{
int flag= 1;
- int error;
- error= setsockopt(server->fd, SOL_SOCKET, SO_KEEPALIVE,
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_KEEPALIVE,
&flag, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
+ assert(error == 0);
}
#ifdef TCP_KEEPIDLE
if (server->root->tcp_keepidle > 0)
{
- int error;
-
- error= setsockopt(server->fd, IPPROTO_TCP, TCP_KEEPIDLE,
- &server->root->tcp_keepidle, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
+ int error= setsockopt(server->fd, IPPROTO_TCP, TCP_KEEPIDLE,
+ &server->root->tcp_keepidle, (socklen_t)sizeof(int));
+ assert(error == 0);
}
#endif
if (server->root->send_size > 0)
{
- int error;
-
- error= setsockopt(server->fd, SOL_SOCKET, SO_SNDBUF,
- &server->root->send_size, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_SNDBUF,
+ &server->root->send_size, (socklen_t)sizeof(int));
+ assert(error == 0);
}
if (server->root->recv_size > 0)
{
- int error;
-
- error= setsockopt(server->fd, SOL_SOCKET, SO_RCVBUF,
- &server->root->recv_size, (socklen_t)sizeof(int));
- WATCHPOINT_ASSERT(error == 0);
+ int error= setsockopt(server->fd, SOL_SOCKET, SO_RCVBUF,
+ &server->root->recv_size, (socklen_t)sizeof(int));
+ assert(error == 0);
}
set_socket_nonblocking(server);
}
-static memcached_return_t unix_socket_connect(memcached_server_st *server)
+static memcached_return_t unix_socket_connect(memcached_instance_st *server)
{
#ifndef WIN32
WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
case EISCONN: /* We were spinning waiting on connect */
{
- WATCHPOINT_ASSERT(0); // Programmer error
+ assert(0); // Programmer error
break;
}
#endif
}
-static memcached_return_t network_connect(memcached_server_st *server)
+static memcached_return_t network_connect(memcached_instance_st *server)
{
bool timeout_error_occured= false;
continue;
}
+ int type= server->address_info_next->ai_socktype;
+ if (HAVE_SOCK_CLOEXEC)
+ {
+ type|= SOCK_CLOEXEC;
+ }
+
if ((server->fd= socket(server->address_info_next->ai_family,
- server->address_info_next->ai_socktype,
+ type,
server->address_info_next->ai_protocol)) < 0)
{
return memcached_set_errno(*server, get_socket_errno(), NULL);
}
+ if (HAVE_SOCK_CLOEXEC == 0)
+ {
+#ifdef FD_CLOEXEC
+ int rval;
+ do
+ {
+ rval= fcntl (server->fd, F_SETFD, FD_CLOEXEC);
+ } while (rval == -1 && (errno == EINTR or errno == EAGAIN));
+#endif
+ }
+
set_socket_options(server);
/* connect to 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
{
if (memcached_has_current_error(*server))
{
- return memcached_server_error_return(server);
+ return memcached_instance_error_return(server);
}
if (timeout_error_occured and server->state < MEMCACHED_SERVER_STATE_IN_PROGRESS)
*/
static memcached_return_t backoff_handling(memcached_server_write_instance_st server, bool& in_timeout)
{
+ struct timeval curr_time;
+ bool _gettime_success= (gettimeofday(&curr_time, NULL) == 0);
+
/*
If we hit server_failure_limit then something is completely wrong about the server.
if (_is_auto_eject_host(server->root))
{
set_last_disconnected_host(server);
- run_distribution((memcached_st *)server->root);
+
+ // 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 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);
}
if (server->state == MEMCACHED_SERVER_STATE_IN_TIMEOUT)
{
- struct timeval curr_time;
- bool _gettime_success= (gettimeofday(&curr_time, NULL) == 0);
-
/*
If next_retry is less then our current time, then we reset and try everything again.
*/
return MEMCACHED_SUCCESS;
}
-memcached_return_t memcached_connect(memcached_server_write_instance_st server)
+static memcached_return_t _memcached_connect(memcached_server_write_instance_st server, const bool set_last_disconnected)
{
if (server->fd != INVALID_SOCKET)
{
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)
{
memcached_mark_server_as_clean(server);
return rc;
}
-
- 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
+ else if (set_last_disconnected)
{
- memcached_set_error(*server, rc, MEMCACHED_AT);
- memcached_mark_server_for_timeout(server);
- }
+ 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)
- {
- return memcached_set_error(*server, MEMCACHED_SERVER_TEMPORARILY_DISABLED, MEMCACHED_AT);
+ 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);
+}