1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2010 Brian Aker All rights reserved.
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are
12 * * Redistributions of source code must retain the above copyright
13 * notice, this list of conditions and the following disclaimer.
15 * * Redistributions in binary form must reproduce the above
16 * copyright notice, this list of conditions and the following disclaimer
17 * in the documentation and/or other materials provided with the
20 * * The names of its contributors may not be used to endorse or
21 * promote products derived from this software without specific prior
24 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
25 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
26 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
27 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
28 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
29 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
30 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
31 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
32 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
33 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
34 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
39 #include <libmemcached/common.h>
44 # define SOCK_CLOEXEC 0
48 # define SOCK_NONBLOCK 0
55 static memcached_return_t
connect_poll(org::libmemcached::Instance
* server
)
58 fds
[0].fd
= server
->fd
;
59 fds
[0].events
= POLLOUT
;
63 if (server
->root
->poll_timeout
== 0)
65 return memcached_set_error(*server
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
68 while (--loop_max
) // Should only loop on cases of ERESTART or EINTR
71 if ((number_of
= poll(fds
, 1, server
->root
->connect_timeout
)) <= 0)
75 int local_errno
= get_socket_errno(); // We cache in case closesocket() modifies errno
78 #ifdef TARGET_OS_LINUX
86 return memcached_set_error(*server
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
89 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"));
91 default: // This should not happen
92 if (fds
[0].revents
& POLLERR
)
95 socklen_t len
= sizeof(err
);
96 if (getsockopt(server
->fd
, SOL_SOCKET
, SO_ERROR
, (char*)&err
, &len
) == 0)
100 // This should never happen, if it does? Punt.
107 assert_msg(server
->fd
!= INVALID_SOCKET
, "poll() was passed an invalid file descriptor");
108 (void)closesocket(server
->fd
);
109 server
->fd
= INVALID_SOCKET
;
110 server
->state
= MEMCACHED_SERVER_STATE_NEW
;
112 return memcached_set_errno(*server
, local_errno
, MEMCACHED_AT
);
115 assert(number_of
== 0);
117 server
->io_wait_count
.timeouts
++;
118 return memcached_set_error(*server
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
121 if (fds
[0].revents
& POLLERR
or
122 fds
[0].revents
& POLLHUP
or
123 fds
[0].revents
& POLLNVAL
)
126 socklen_t len
= sizeof (err
);
127 if (getsockopt(fds
[0].fd
, SOL_SOCKET
, SO_ERROR
, (char*)&err
, &len
) == 0)
129 // We check the value to see what happened wth the socket.
132 return MEMCACHED_SUCCESS
;
137 return memcached_set_errno(*server
, err
, MEMCACHED_AT
);
139 assert(fds
[0].revents
& POLLIN
or fds
[0].revents
& POLLOUT
);
141 return MEMCACHED_SUCCESS
;
144 // This should only be possible from ERESTART or EINTR;
145 return memcached_set_errno(*server
, get_socket_errno(), MEMCACHED_AT
);
148 static memcached_return_t
set_hostinfo(org::libmemcached::Instance
* server
)
150 assert(server
->type
!= MEMCACHED_CONNECTION_UNIX_SOCKET
);
151 if (server
->address_info
)
153 freeaddrinfo(server
->address_info
);
154 server
->address_info
= NULL
;
155 server
->address_info_next
= NULL
;
158 char str_port
[MEMCACHED_NI_MAXSERV
];
159 int length
= snprintf(str_port
, MEMCACHED_NI_MAXSERV
, "%u", uint32_t(server
->port()));
160 if (length
>= MEMCACHED_NI_MAXSERV
or length
<= 0)
162 return memcached_set_error(*server
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
,
163 memcached_literal_param("snprintf(NI_MAXSERV)"));
166 struct addrinfo hints
;
167 memset(&hints
, 0, sizeof(struct addrinfo
));
170 hints
.ai_family
= AF_INET
;
172 if (memcached_is_udp(server
->root
))
174 hints
.ai_protocol
= IPPROTO_UDP
;
175 hints
.ai_socktype
= SOCK_DGRAM
;
179 hints
.ai_socktype
= SOCK_STREAM
;
180 hints
.ai_protocol
= IPPROTO_TCP
;
183 assert(server
->address_info
== NULL
);
184 assert(server
->address_info_next
== NULL
);
186 switch(errcode
= getaddrinfo(server
->hostname
, str_port
, &hints
, &server
->address_info
))
192 if (server
->address_info
)
194 freeaddrinfo(server
->address_info
);
195 server
->address_info
= NULL
;
196 server
->address_info_next
= NULL
;
198 return memcached_set_error(*server
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
, memcached_string_make_from_cstr(gai_strerror(errcode
)));
201 if (server
->address_info
)
203 freeaddrinfo(server
->address_info
);
204 server
->address_info
= NULL
;
205 server
->address_info_next
= NULL
;
207 return memcached_set_errno(*server
, errno
, MEMCACHED_AT
, memcached_literal_param("getaddrinfo(EAI_SYSTEM)"));
210 if (server
->address_info
)
212 freeaddrinfo(server
->address_info
);
213 server
->address_info
= NULL
;
214 server
->address_info_next
= NULL
;
216 return memcached_set_error(*server
, MEMCACHED_INVALID_ARGUMENTS
, MEMCACHED_AT
, memcached_literal_param("getaddrinfo(EAI_BADFLAGS)"));
219 if (server
->address_info
)
221 freeaddrinfo(server
->address_info
);
222 server
->address_info
= NULL
;
223 server
->address_info_next
= NULL
;
225 return memcached_set_error(*server
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
229 if (server
->address_info
)
231 freeaddrinfo(server
->address_info
);
232 server
->address_info
= NULL
;
233 server
->address_info_next
= NULL
;
235 return memcached_set_error(*server
, MEMCACHED_HOST_LOOKUP_FAILURE
, MEMCACHED_AT
, memcached_string_make_from_cstr(gai_strerror(errcode
)));
238 server
->address_info_next
= server
->address_info
;
239 server
->state
= MEMCACHED_SERVER_STATE_ADDRINFO
;
241 return MEMCACHED_SUCCESS
;
244 static inline void set_socket_nonblocking(org::libmemcached::Instance
* server
)
248 if (ioctlsocket(server
->fd
, FIONBIO
, &arg
) == SOCKET_ERROR
)
250 memcached_set_errno(*server
, get_socket_errno(), NULL
);
255 if (SOCK_NONBLOCK
== 0)
259 flags
= fcntl(server
->fd
, F_GETFL
, 0);
260 } while (flags
== -1 && (errno
== EINTR
|| errno
== EAGAIN
));
264 memcached_set_errno(*server
, errno
, NULL
);
266 else if ((flags
& O_NONBLOCK
) == 0)
272 rval
= fcntl(server
->fd
, F_SETFL
, flags
| O_NONBLOCK
);
273 } while (rval
== -1 && (errno
== EINTR
or errno
== EAGAIN
));
277 memcached_set_errno(*server
, errno
, NULL
);
284 static void set_socket_options(org::libmemcached::Instance
* server
)
286 assert_msg(server
->fd
!= INVALID_SOCKET
, "invalid socket was passed to set_socket_options()");
288 if (memcached_is_udp(server
->root
))
294 if (server
->root
->snd_timeout
> 0)
296 struct timeval waittime
;
298 waittime
.tv_sec
= server
->root
->snd_timeout
/ 1000000;
299 waittime
.tv_usec
= server
->root
->snd_timeout
% 1000000;
301 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_SNDTIMEO
,
302 (char*)&waittime
, (socklen_t
)sizeof(struct timeval
));
309 if (server
->root
->rcv_timeout
> 0)
311 struct timeval waittime
;
313 waittime
.tv_sec
= server
->root
->rcv_timeout
/ 1000000;
314 waittime
.tv_usec
= server
->root
->rcv_timeout
% 1000000;
316 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_RCVTIMEO
,
317 (char*)&waittime
, (socklen_t
)sizeof(struct timeval
));
324 #if defined(__MACH__) && defined(__APPLE__) || defined(__FreeBSD__)
327 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_NOSIGPIPE
, (void *)&set
, sizeof(int));
331 // This is not considered a fatal error
335 perror("setsockopt(SO_NOSIGPIPE)");
341 if (server
->root
->flags
.no_block
)
343 struct linger linger
;
346 linger
.l_linger
= 0; /* By default on close() just drop the socket */
347 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_LINGER
,
348 (char*)&linger
, (socklen_t
)sizeof(struct linger
));
353 if (server
->root
->flags
.tcp_nodelay
)
357 int error
= setsockopt(server
->fd
, IPPROTO_TCP
, TCP_NODELAY
,
358 (char*)&flag
, (socklen_t
)sizeof(int));
363 if (server
->root
->flags
.tcp_keepalive
)
367 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_KEEPALIVE
,
368 (char*)&flag
, (socklen_t
)sizeof(int));
374 if (server
->root
->tcp_keepidle
> 0)
376 int error
= setsockopt(server
->fd
, IPPROTO_TCP
, TCP_KEEPIDLE
,
377 (char*)&server
->root
->tcp_keepidle
, (socklen_t
)sizeof(int));
383 if (server
->root
->send_size
> 0)
385 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_SNDBUF
,
386 (char*)&server
->root
->send_size
, (socklen_t
)sizeof(int));
391 if (server
->root
->recv_size
> 0)
393 int error
= setsockopt(server
->fd
, SOL_SOCKET
, SO_RCVBUF
,
394 (char*)&server
->root
->recv_size
, (socklen_t
)sizeof(int));
399 /* libmemcached will always use nonblocking IO to avoid write deadlocks */
400 set_socket_nonblocking(server
);
403 static memcached_return_t
unix_socket_connect(org::libmemcached::Instance
* server
)
406 WATCHPOINT_ASSERT(server
->fd
== INVALID_SOCKET
);
408 int type
= SOCK_STREAM
;
416 type
|= SOCK_NONBLOCK
;
419 if ((server
->fd
= socket(AF_UNIX
, type
, 0)) < 0)
421 memcached_set_errno(*server
, errno
, NULL
);
422 return MEMCACHED_CONNECTION_FAILURE
;
425 struct sockaddr_un servAddr
;
427 memset(&servAddr
, 0, sizeof (struct sockaddr_un
));
428 servAddr
.sun_family
= AF_UNIX
;
429 strncpy(servAddr
.sun_path
, server
->hostname
, sizeof(servAddr
.sun_path
)); /* Copy filename */
432 if (connect(server
->fd
, (struct sockaddr
*)&servAddr
, sizeof(servAddr
)) < 0)
441 case EISCONN
: /* We were spinning waiting on connect */
443 assert(0); // Programmer error
448 WATCHPOINT_ERRNO(errno
);
449 memcached_set_errno(*server
, errno
, MEMCACHED_AT
);
450 return MEMCACHED_CONNECTION_FAILURE
;
454 server
->state
= MEMCACHED_SERVER_STATE_CONNECTED
;
456 WATCHPOINT_ASSERT(server
->fd
!= INVALID_SOCKET
);
458 return MEMCACHED_SUCCESS
;
461 return MEMCACHED_NOT_SUPPORTED
;
465 static memcached_return_t
network_connect(org::libmemcached::Instance
* server
)
467 bool timeout_error_occured
= false;
469 WATCHPOINT_ASSERT(server
->fd
== INVALID_SOCKET
);
470 WATCHPOINT_ASSERT(server
->cursor_active_
== 0);
473 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.
475 if (server
->address_info
== NULL
or server
->address_info_next
== NULL
)
477 WATCHPOINT_ASSERT(server
->state
== MEMCACHED_SERVER_STATE_NEW
);
478 server
->address_info_next
= NULL
;
479 memcached_return_t rc
= set_hostinfo(server
);
481 if (memcached_failed(rc
))
487 if (server
->address_info_next
== NULL
)
489 server
->address_info_next
= server
->address_info
;
490 server
->state
= MEMCACHED_SERVER_STATE_ADDRINFO
;
493 /* Create the socket */
494 while (server
->address_info_next
and server
->fd
== INVALID_SOCKET
)
496 /* Memcache server does not support IPV6 in udp mode, so skip if not ipv4 */
497 if (memcached_is_udp(server
->root
) and server
->address_info_next
->ai_family
!= AF_INET
)
499 server
->address_info_next
= server
->address_info_next
->ai_next
;
503 int type
= server
->address_info_next
->ai_socktype
;
511 type
|= SOCK_NONBLOCK
;
514 server
->fd
= socket(server
->address_info_next
->ai_family
,
516 server
->address_info_next
->ai_protocol
);
518 if (int(server
->fd
) == SOCKET_ERROR
)
520 return memcached_set_errno(*server
, get_socket_errno(), NULL
);
524 // If SOCK_CLOEXEC exists then we don't need to call the following
525 if (SOCK_CLOEXEC
== 0)
532 rval
= fcntl (server
->fd
, F_SETFD
, FD_CLOEXEC
);
533 } while (rval
== -1 && (errno
== EINTR
or errno
== EAGAIN
));
538 set_socket_options(server
);
540 /* connect to server */
541 if ((connect(server
->fd
, server
->address_info_next
->ai_addr
, server
->address_info_next
->ai_addrlen
) != SOCKET_ERROR
))
543 server
->state
= MEMCACHED_SERVER_STATE_CONNECTED
;
544 return MEMCACHED_SUCCESS
;
547 /* An error occurred */
548 switch (get_socket_errno())
551 timeout_error_occured
= true;
555 #if EWOULDBLOCK != EAGAIN
558 case EINPROGRESS
: // nonblocking mode - first return
559 case EALREADY
: // nonblocking mode - subsequent returns
561 server
->state
= MEMCACHED_SERVER_STATE_IN_PROGRESS
;
562 memcached_return_t rc
= connect_poll(server
);
564 if (memcached_success(rc
))
566 server
->state
= MEMCACHED_SERVER_STATE_CONNECTED
;
567 return MEMCACHED_SUCCESS
;
570 // A timeout here is treated as an error, we will not retry
571 if (rc
== MEMCACHED_TIMEOUT
)
573 timeout_error_occured
= true;
578 case EISCONN
: // we are connected :-)
579 WATCHPOINT_ASSERT(0); // This is a programmer's error
582 case EINTR
: // Special case, we retry ai_addr
583 WATCHPOINT_ASSERT(server
->fd
!= INVALID_SOCKET
);
584 (void)closesocket(server
->fd
);
585 server
->fd
= INVALID_SOCKET
;
592 WATCHPOINT_ASSERT(server
->fd
!= INVALID_SOCKET
);
593 (void)closesocket(server
->fd
);
594 server
->fd
= INVALID_SOCKET
;
595 server
->address_info_next
= server
->address_info_next
->ai_next
;
598 WATCHPOINT_ASSERT(server
->fd
== INVALID_SOCKET
);
600 if (timeout_error_occured
)
602 if (server
->fd
!= INVALID_SOCKET
)
604 (void)closesocket(server
->fd
);
605 server
->fd
= INVALID_SOCKET
;
609 WATCHPOINT_STRING("Never got a good file descriptor");
611 if (memcached_has_current_error(*server
))
613 return memcached_instance_error_return(server
);
616 if (timeout_error_occured
and server
->state
< MEMCACHED_SERVER_STATE_IN_PROGRESS
)
618 return memcached_set_error(*server
, MEMCACHED_TIMEOUT
, MEMCACHED_AT
);
621 return memcached_set_error(*server
, MEMCACHED_CONNECTION_FAILURE
, MEMCACHED_AT
); /* The last error should be from connect() */
628 Based on time/failure count fail the connect without trying. This prevents waiting in a state where
629 we get caught spending cycles just waiting.
631 static memcached_return_t
backoff_handling(org::libmemcached::Instance
* server
, bool& in_timeout
)
633 struct timeval curr_time
;
634 bool _gettime_success
= (gettimeofday(&curr_time
, NULL
) == 0);
637 If we hit server_failure_limit then something is completely wrong about the server.
639 1) If autoeject is enabled we do that.
640 2) If not? We go into timeout again, there is much else to do :(
642 if (server
->server_failure_counter
>= server
->root
->server_failure_limit
)
645 We just auto_eject if we hit this point
647 if (_is_auto_eject_host(server
->root
))
649 set_last_disconnected_host(server
);
651 // Retry dead servers if requested
652 if (_gettime_success
and server
->root
->dead_timeout
> 0)
654 server
->next_retry
= curr_time
.tv_sec
+server
->root
->dead_timeout
;
656 // We only retry dead servers once before assuming failure again
657 server
->server_failure_counter
= server
->root
->server_failure_limit
-1;
660 memcached_return_t rc
;
661 if (memcached_failed(rc
= run_distribution((memcached_st
*)server
->root
)))
663 return memcached_set_error(*server
, rc
, MEMCACHED_AT
, memcached_literal_param("Backoff handling failed during run_distribution"));
666 return memcached_set_error(*server
, MEMCACHED_SERVER_MARKED_DEAD
, MEMCACHED_AT
);
669 server
->state
= MEMCACHED_SERVER_STATE_IN_TIMEOUT
;
671 // Sanity check/setting
672 if (server
->next_retry
== 0)
674 server
->next_retry
= 1;
678 if (server
->state
== MEMCACHED_SERVER_STATE_IN_TIMEOUT
)
681 If next_retry is less then our current time, then we reset and try everything again.
683 if (_gettime_success
and server
->next_retry
< curr_time
.tv_sec
)
685 server
->state
= MEMCACHED_SERVER_STATE_NEW
;
689 return memcached_set_error(*server
, MEMCACHED_SERVER_TEMPORARILY_DISABLED
, MEMCACHED_AT
);
695 return MEMCACHED_SUCCESS
;
698 static memcached_return_t
_memcached_connect(org::libmemcached::Instance
* server
, const bool set_last_disconnected
)
701 if (server
->fd
!= INVALID_SOCKET
)
703 return MEMCACHED_SUCCESS
;
706 LIBMEMCACHED_MEMCACHED_CONNECT_START();
708 bool in_timeout
= false;
709 memcached_return_t rc
;
710 if (memcached_failed(rc
= backoff_handling(server
, in_timeout
)))
712 set_last_disconnected_host(server
);
716 if (LIBMEMCACHED_WITH_SASL_SUPPORT
and server
->root
->sasl
.callbacks
and memcached_is_udp(server
->root
))
718 return memcached_set_error(*server
, MEMCACHED_INVALID_HOST_PROTOCOL
, MEMCACHED_AT
, memcached_literal_param("SASL is not supported for UDP connections"));
721 if (server
->hostname
[0] == '/')
723 server
->type
= MEMCACHED_CONNECTION_UNIX_SOCKET
;
726 /* We need to clean up the multi startup piece */
727 switch (server
->type
)
729 case MEMCACHED_CONNECTION_UDP
:
730 case MEMCACHED_CONNECTION_TCP
:
731 rc
= network_connect(server
);
733 if (LIBMEMCACHED_WITH_SASL_SUPPORT
)
735 if (server
->fd
!= INVALID_SOCKET
and server
->root
->sasl
.callbacks
)
737 rc
= memcached_sasl_authenticate_connection(server
);
738 fprintf(stderr
, "%s:%d %s\n", __FILE__
, __LINE__
, memcached_strerror(NULL
, rc
));
739 if (memcached_failed(rc
) and server
->fd
!= INVALID_SOCKET
)
741 WATCHPOINT_ASSERT(server
->fd
!= INVALID_SOCKET
);
742 (void)closesocket(server
->fd
);
743 server
->fd
= INVALID_SOCKET
;
749 case MEMCACHED_CONNECTION_UNIX_SOCKET
:
750 rc
= unix_socket_connect(server
);
754 if (memcached_success(rc
))
756 server
->mark_server_as_clean();
757 memcached_version_instance(server
);
760 else if (set_last_disconnected
)
762 set_last_disconnected_host(server
);
763 if (memcached_has_current_error(*server
))
765 memcached_mark_server_for_timeout(server
);
766 assert(memcached_failed(memcached_instance_error_return(server
)));
770 memcached_set_error(*server
, rc
, MEMCACHED_AT
);
771 memcached_mark_server_for_timeout(server
);
774 LIBMEMCACHED_MEMCACHED_CONNECT_END();
779 int snprintf_length
= snprintf(buffer
, sizeof(buffer
), "%s:%d", server
->hostname
, int(server
->port()));
780 return memcached_set_error(*server
, MEMCACHED_SERVER_TEMPORARILY_DISABLED
, MEMCACHED_AT
, buffer
, snprintf_length
);
787 memcached_return_t
memcached_connect_try(org::libmemcached::Instance
* server
)
789 if (server
and server
->root
and server
->root
->state
.is_parsing
)
791 return MEMCACHED_SUCCESS
;
794 return _memcached_connect(server
, false);
797 memcached_return_t
memcached_connect(org::libmemcached::Instance
* server
)
799 return _memcached_connect(server
, true);