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