Update libtest/fedora/lion fixes
[m6w6/libmemcached] / libmemcached / connect.cc
1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
2 *
3 * Libmemcached library
4 *
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2010 Brian Aker All rights reserved.
7 *
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are
10 * met:
11 *
12 * * Redistributions of source code must retain the above copyright
13 * notice, this list of conditions and the following disclaimer.
14 *
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
18 * distribution.
19 *
20 * * The names of its contributors may not be used to endorse or
21 * promote products derived from this software without specific prior
22 * written permission.
23 *
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.
35 *
36 */
37
38
39 #include <libmemcached/common.h>
40 #include <ctime>
41 #include <sys/time.h>
42
43 static memcached_return_t connect_poll(memcached_server_st *ptr)
44 {
45 struct pollfd fds[1];
46 fds[0].fd = ptr->fd;
47 fds[0].events = POLLOUT;
48
49 size_t loop_max= 5;
50
51 if (ptr->root->poll_timeout == 0)
52 {
53 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
54 }
55
56 while (--loop_max) // Should only loop on cases of ERESTART or EINTR
57 {
58 int error= poll(fds, 1, ptr->root->connect_timeout);
59 switch (error)
60 {
61 case 1:
62 {
63 int err;
64 socklen_t len= sizeof (err);
65 (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len);
66
67 // We check the value to see what happened wth the socket.
68 if (err == 0)
69 {
70 return MEMCACHED_SUCCESS;
71 }
72
73 return memcached_set_errno(*ptr, err, MEMCACHED_AT);
74 }
75 case 0:
76 {
77 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
78 }
79
80 default: // A real error occurred and we need to completely bail
81 switch (get_socket_errno())
82 {
83 #ifdef TARGET_OS_LINUX
84 case ERESTART:
85 #endif
86 case EINTR:
87 continue;
88
89 case EFAULT:
90 case ENOMEM:
91 return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
92
93 case EINVAL:
94 return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid"));
95
96 default: // This should not happen
97 if (fds[0].revents & POLLERR)
98 {
99 int err;
100 socklen_t len= sizeof (err);
101 (void)getsockopt(ptr->fd, SOL_SOCKET, SO_ERROR, &err, &len);
102 memcached_set_errno(*ptr, (err == 0) ? get_socket_errno() : err, MEMCACHED_AT);
103 }
104 else
105 {
106 memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
107 }
108
109 assert_msg(ptr->fd != INVALID_SOCKET, "poll() was passed an invalid file descriptor");
110 (void)closesocket(ptr->fd);
111 ptr->fd= INVALID_SOCKET;
112 ptr->state= MEMCACHED_SERVER_STATE_NEW;
113
114 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
115 }
116 }
117 }
118
119 // This should only be possible from ERESTART or EINTR;
120 return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
121 }
122
123 static memcached_return_t set_hostinfo(memcached_server_st *server)
124 {
125 if (server->address_info)
126 {
127 freeaddrinfo(server->address_info);
128 server->address_info= NULL;
129 server->address_info_next= NULL;
130 }
131
132 char str_port[NI_MAXSERV];
133 int length= snprintf(str_port, NI_MAXSERV, "%u", (uint32_t)server->port);
134 if (length >= NI_MAXSERV or length < 0)
135 {
136 return MEMCACHED_FAILURE;
137 }
138
139 struct addrinfo hints;
140 memset(&hints, 0, sizeof(struct addrinfo));
141
142 #if 0
143 hints.ai_family= AF_INET;
144 #endif
145 if (server->type == MEMCACHED_CONNECTION_UDP)
146 {
147 hints.ai_protocol= IPPROTO_UDP;
148 hints.ai_socktype= SOCK_DGRAM;
149 }
150 else
151 {
152 hints.ai_socktype= SOCK_STREAM;
153 hints.ai_protocol= IPPROTO_TCP;
154 }
155
156 int errcode;
157 switch(errcode= getaddrinfo(server->hostname, str_port, &hints, &server->address_info))
158 {
159 case 0:
160 break;
161
162 case EAI_AGAIN:
163 return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
164
165 case EAI_SYSTEM:
166 return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_SYSTEM)"));
167
168 case EAI_BADFLAGS:
169 return memcached_set_error(*server, MEMCACHED_INVALID_ARGUMENTS, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_BADFLAGS)"));
170
171 case EAI_MEMORY:
172 return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
173
174 default:
175 {
176 return memcached_set_error(*server, MEMCACHED_HOST_LOOKUP_FAILURE, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
177 }
178 }
179 server->address_info_next= server->address_info;
180 server->state= MEMCACHED_SERVER_STATE_ADDRINFO;
181
182 return MEMCACHED_SUCCESS;
183 }
184
185 static inline void set_socket_nonblocking(memcached_server_st *ptr)
186 {
187 #ifdef WIN32
188 u_long arg = 1;
189 if (ioctlsocket(ptr->fd, FIONBIO, &arg) == SOCKET_ERROR)
190 {
191 memcached_set_errno(*ptr, get_socket_errno(), NULL);
192 }
193 #else
194 int flags;
195
196 do
197 {
198 flags= fcntl(ptr->fd, F_GETFL, 0);
199 } while (flags == -1 && (errno == EINTR || errno == EAGAIN));
200
201 if (flags == -1)
202 {
203 memcached_set_errno(*ptr, errno, NULL);
204 }
205 else if ((flags & O_NONBLOCK) == 0)
206 {
207 int rval;
208
209 do
210 {
211 rval= fcntl(ptr->fd, F_SETFL, flags | O_NONBLOCK);
212 } while (rval == -1 && (errno == EINTR || errno == EAGAIN));
213
214 unlikely (rval == -1)
215 {
216 memcached_set_errno(*ptr, errno, NULL);
217 }
218 }
219 #endif
220 }
221
222 static void set_socket_options(memcached_server_st *ptr)
223 {
224 assert_msg(ptr->fd != -1, "invalid socket was passed to set_socket_options()");
225
226 if (ptr->type == MEMCACHED_CONNECTION_UDP)
227 {
228 return;
229 }
230
231 #ifdef HAVE_SNDTIMEO
232 if (ptr->root->snd_timeout)
233 {
234 int error;
235 struct timeval waittime;
236
237 waittime.tv_sec= 0;
238 waittime.tv_usec= ptr->root->snd_timeout;
239
240 error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDTIMEO,
241 &waittime, (socklen_t)sizeof(struct timeval));
242 WATCHPOINT_ASSERT(error == 0);
243 }
244 #endif
245
246 #ifdef HAVE_RCVTIMEO
247 if (ptr->root->rcv_timeout)
248 {
249 int error;
250 struct timeval waittime;
251
252 waittime.tv_sec= 0;
253 waittime.tv_usec= ptr->root->rcv_timeout;
254
255 error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVTIMEO,
256 &waittime, (socklen_t)sizeof(struct timeval));
257 WATCHPOINT_ASSERT(error == 0);
258 }
259 #endif
260
261
262 #if defined(__MACH__) && defined(__APPLE__) || defined(__FreeBSD__)
263 {
264 int set= 1;
265 int error= setsockopt(ptr->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
266
267 // This is not considered a fatal error
268 if (error == -1)
269 {
270 WATCHPOINT_ERRNO(get_socket_errno());
271 perror("setsockopt(SO_NOSIGPIPE)");
272 }
273 }
274 #endif
275
276 if (ptr->root->flags.no_block)
277 {
278 int error;
279 struct linger linger;
280
281 linger.l_onoff= 1;
282 linger.l_linger= 0; /* By default on close() just drop the socket */
283 error= setsockopt(ptr->fd, SOL_SOCKET, SO_LINGER,
284 &linger, (socklen_t)sizeof(struct linger));
285 WATCHPOINT_ASSERT(error == 0);
286 }
287
288 if (ptr->root->flags.tcp_nodelay)
289 {
290 int flag= 1;
291 int error;
292
293 error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_NODELAY,
294 &flag, (socklen_t)sizeof(int));
295 WATCHPOINT_ASSERT(error == 0);
296 }
297
298 if (ptr->root->flags.tcp_keepalive)
299 {
300 int flag= 1;
301 int error;
302
303 error= setsockopt(ptr->fd, SOL_SOCKET, SO_KEEPALIVE,
304 &flag, (socklen_t)sizeof(int));
305 WATCHPOINT_ASSERT(error == 0);
306 }
307
308 #ifdef TCP_KEEPIDLE
309 if (ptr->root->tcp_keepidle > 0)
310 {
311 int error;
312
313 error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_KEEPIDLE,
314 &ptr->root->tcp_keepidle, (socklen_t)sizeof(int));
315 WATCHPOINT_ASSERT(error == 0);
316 }
317 #endif
318
319 if (ptr->root->send_size > 0)
320 {
321 int error;
322
323 error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF,
324 &ptr->root->send_size, (socklen_t)sizeof(int));
325 WATCHPOINT_ASSERT(error == 0);
326 }
327
328 if (ptr->root->recv_size > 0)
329 {
330 int error;
331
332 error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVBUF,
333 &ptr->root->recv_size, (socklen_t)sizeof(int));
334 WATCHPOINT_ASSERT(error == 0);
335 }
336
337
338 /* libmemcached will always use nonblocking IO to avoid write deadlocks */
339 set_socket_nonblocking(ptr);
340 }
341
342 static memcached_return_t unix_socket_connect(memcached_server_st *ptr)
343 {
344 #ifndef WIN32
345 WATCHPOINT_ASSERT(ptr->fd == -1);
346
347 if ((ptr->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
348 {
349 memcached_set_errno(*ptr, errno, NULL);
350 return MEMCACHED_CONNECTION_FAILURE;
351 }
352
353 struct sockaddr_un servAddr;
354
355 memset(&servAddr, 0, sizeof (struct sockaddr_un));
356 servAddr.sun_family= AF_UNIX;
357 strncpy(servAddr.sun_path, ptr->hostname, sizeof(servAddr.sun_path)); /* Copy filename */
358
359 do {
360 if (connect(ptr->fd, (struct sockaddr *)&servAddr, sizeof(servAddr)) < 0)
361 {
362 switch (errno)
363 {
364 case EINPROGRESS:
365 case EALREADY:
366 case EINTR:
367 continue;
368
369 case EISCONN: /* We were spinning waiting on connect */
370 {
371 WATCHPOINT_ASSERT(0); // Programmer error
372 break;
373 }
374
375 default:
376 WATCHPOINT_ERRNO(errno);
377 memcached_set_errno(*ptr, errno, MEMCACHED_AT);
378 return MEMCACHED_CONNECTION_FAILURE;
379 }
380 }
381 } while (0);
382 ptr->state= MEMCACHED_SERVER_STATE_CONNECTED;
383
384 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
385
386 return MEMCACHED_SUCCESS;
387 #else
388 (void)ptr;
389 return MEMCACHED_NOT_SUPPORTED;
390 #endif
391 }
392
393 static memcached_return_t network_connect(memcached_server_st *ptr)
394 {
395 bool timeout_error_occured= false;
396
397 WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
398 WATCHPOINT_ASSERT(ptr->cursor_active == 0);
399
400 if (not ptr->address_info)
401 {
402 WATCHPOINT_ASSERT(ptr->state == MEMCACHED_SERVER_STATE_NEW);
403 memcached_return_t rc;
404 uint32_t counter= 5;
405 while (--counter)
406 {
407 if ((rc= set_hostinfo(ptr)) != MEMCACHED_TIMEOUT)
408 {
409 break;
410 }
411
412 #ifndef WIN32
413 struct timespec dream, rem;
414
415 dream.tv_nsec= 1000;
416 dream.tv_sec= 0;
417
418 nanosleep(&dream, &rem);
419 #endif
420 }
421
422 if (memcached_failed(rc))
423 return rc;
424 }
425
426 /* Create the socket */
427 while (ptr->address_info_next && ptr->fd == INVALID_SOCKET)
428 {
429 /* Memcache server does not support IPV6 in udp mode, so skip if not ipv4 */
430 if (ptr->type == MEMCACHED_CONNECTION_UDP && ptr->address_info_next->ai_family != AF_INET)
431 {
432 ptr->address_info_next= ptr->address_info_next->ai_next;
433 continue;
434 }
435
436 if ((ptr->fd= socket(ptr->address_info_next->ai_family,
437 ptr->address_info_next->ai_socktype,
438 ptr->address_info_next->ai_protocol)) < 0)
439 {
440 return memcached_set_errno(*ptr, get_socket_errno(), NULL);
441 }
442
443 set_socket_options(ptr);
444
445 /* connect to server */
446 if ((connect(ptr->fd, ptr->address_info_next->ai_addr, ptr->address_info_next->ai_addrlen) != SOCKET_ERROR))
447 {
448 ptr->state= MEMCACHED_SERVER_STATE_CONNECTED;
449 break; // Success
450 }
451
452 /* An error occurred */
453 switch (get_socket_errno())
454 {
455 case ETIMEDOUT:
456 timeout_error_occured= true;
457 break;
458
459 case EWOULDBLOCK:
460 case EINPROGRESS: // nonblocking mode - first return
461 case EALREADY: // nonblocking mode - subsequent returns
462 {
463 ptr->state= MEMCACHED_SERVER_STATE_IN_PROGRESS;
464 memcached_return_t rc= connect_poll(ptr);
465
466 if (memcached_success(rc))
467 {
468 ptr->state= MEMCACHED_SERVER_STATE_CONNECTED;
469 return MEMCACHED_SUCCESS;
470 }
471
472 // A timeout here is treated as an error, we will not retry
473 if (rc == MEMCACHED_TIMEOUT)
474 {
475 timeout_error_occured= true;
476 }
477 }
478 break;
479
480 case EISCONN: // we are connected :-)
481 WATCHPOINT_ASSERT(0); // This is a programmer's error
482 break;
483
484 case EINTR: // Special case, we retry ai_addr
485 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
486 (void)closesocket(ptr->fd);
487 ptr->fd= INVALID_SOCKET;
488 continue;
489
490 default:
491 break;
492 }
493
494 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
495 (void)closesocket(ptr->fd);
496 ptr->fd= INVALID_SOCKET;
497 ptr->address_info_next= ptr->address_info_next->ai_next;
498 }
499
500 WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
501
502 if (timeout_error_occured)
503 {
504 if (ptr->fd != INVALID_SOCKET)
505 {
506 (void)closesocket(ptr->fd);
507 ptr->fd= INVALID_SOCKET;
508 }
509 }
510
511 WATCHPOINT_STRING("Never got a good file descriptor");
512 /* Failed to connect. schedule next retry */
513 if (ptr->root->retry_timeout)
514 {
515 struct timeval next_time;
516
517 if (gettimeofday(&next_time, NULL) == 0)
518 {
519 ptr->next_retry= next_time.tv_sec + ptr->root->retry_timeout;
520 }
521 }
522
523 if (memcached_has_current_error(*ptr))
524 {
525 return memcached_server_error_return(ptr);
526 }
527
528 if (timeout_error_occured and ptr->state < MEMCACHED_SERVER_STATE_IN_PROGRESS)
529 {
530 return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
531 }
532
533 return memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); /* The last error should be from connect() */
534 }
535
536 void set_last_disconnected_host(memcached_server_write_instance_st self)
537 {
538 // const_cast
539 memcached_st *root= (memcached_st *)self->root;
540
541 memcached_server_free(root->last_disconnected_server);
542 root->last_disconnected_server= memcached_server_clone(NULL, self);
543 }
544
545 memcached_return_t memcached_connect(memcached_server_write_instance_st ptr)
546 {
547 memcached_return_t rc= MEMCACHED_NO_SERVERS;
548
549 if (ptr->fd != INVALID_SOCKET)
550 {
551 return MEMCACHED_SUCCESS;
552 }
553
554 LIBMEMCACHED_MEMCACHED_CONNECT_START();
555
556 /* both retry_timeout and server_failure_limit must be set in order to delay retrying a server on error. */
557 WATCHPOINT_ASSERT(ptr->root);
558 if (ptr->root->retry_timeout && ptr->next_retry)
559 {
560 struct timeval curr_time;
561
562 gettimeofday(&curr_time, NULL);
563
564 // We should optimize this to remove the allocation if the server was
565 // the last server to die
566 if (ptr->next_retry > curr_time.tv_sec)
567 {
568 set_last_disconnected_host(ptr);
569
570 return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
571 }
572 }
573
574 // If we are over the counter failure, we just fail. Reject host only
575 // works if you have a set number of failures.
576 if (ptr->root->server_failure_limit && ptr->server_failure_counter >= ptr->root->server_failure_limit)
577 {
578 set_last_disconnected_host(ptr);
579
580 // @todo fix this by fixing behavior to no longer make use of
581 // memcached_st
582 if (_is_auto_eject_host(ptr->root))
583 {
584 run_distribution((memcached_st *)ptr->root);
585 }
586
587 return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
588 }
589
590 /* We need to clean up the multi startup piece */
591 switch (ptr->type)
592 {
593 case MEMCACHED_CONNECTION_UNKNOWN:
594 WATCHPOINT_ASSERT(0);
595 rc= MEMCACHED_NOT_SUPPORTED;
596 break;
597
598 case MEMCACHED_CONNECTION_UDP:
599 case MEMCACHED_CONNECTION_TCP:
600 rc= network_connect(ptr);
601 #ifdef LIBMEMCACHED_WITH_SASL_SUPPORT
602 if (ptr->fd != INVALID_SOCKET and ptr->root->sasl.callbacks)
603 {
604 rc= memcached_sasl_authenticate_connection(ptr);
605 if (memcached_failed(rc) and ptr->fd != INVALID_SOCKET)
606 {
607 WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
608 (void)closesocket(ptr->fd);
609 ptr->fd= INVALID_SOCKET;
610 }
611 }
612 #endif
613 break;
614
615 case MEMCACHED_CONNECTION_UNIX_SOCKET:
616 rc= unix_socket_connect(ptr);
617 break;
618
619 case MEMCACHED_CONNECTION_MAX:
620 default:
621 WATCHPOINT_ASSERT(0);
622 }
623
624 if (memcached_success(rc))
625 {
626 ptr->server_failure_counter= 0;
627 ptr->next_retry= 0;
628 }
629 else if (memcached_has_current_error(*ptr))
630 {
631 ptr->server_failure_counter++;
632 set_last_disconnected_host(ptr);
633 }
634 else
635 {
636 memcached_set_error(*ptr, rc, MEMCACHED_AT);
637 ptr->server_failure_counter++;
638 set_last_disconnected_host(ptr);
639 }
640
641 LIBMEMCACHED_MEMCACHED_CONNECT_END();
642
643 return rc;
644 }