Test cleanup, extended error system to record host.
[awesomized/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 <assert.h>
41 #include <sys/time.h>
42 #include <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 uint32_t counter= 5;
149 while (--counter)
150 {
151 int e= getaddrinfo(server->hostname, str_port, &hints, &server->address_info);
152
153 if (e == 0)
154 {
155 break;
156 }
157 else if (e == EAI_AGAIN)
158 {
159 #ifndef WIN32
160 struct timespec dream, rem;
161
162 dream.tv_nsec= 1000;
163 dream.tv_sec= 0;
164
165 nanosleep(&dream, &rem);
166 #endif
167 continue;
168 }
169 else
170 {
171 WATCHPOINT_STRING(server->hostname);
172 WATCHPOINT_STRING(gai_strerror(e));
173 return MEMCACHED_HOST_LOOKUP_FAILURE;
174 }
175 }
176
177 server->address_info_next= server->address_info;
178
179 return MEMCACHED_SUCCESS;
180 }
181
182 static inline memcached_return_t set_socket_nonblocking(memcached_server_st *ptr)
183 {
184 #ifdef WIN32
185 u_long arg = 1;
186 if (ioctlsocket(ptr->fd, FIONBIO, &arg) == SOCKET_ERROR)
187 {
188 return memcached_set_errno(*ptr, get_socket_errno(), NULL);
189 }
190 #else
191 int flags;
192
193 do
194 {
195 flags= fcntl(ptr->fd, F_GETFL, 0);
196 }
197 while (flags == -1 && (errno == EINTR || errno == EAGAIN));
198
199 unlikely (flags == -1)
200 {
201 return memcached_set_errno(*ptr, errno, NULL);
202 }
203 else if ((flags & O_NONBLOCK) == 0)
204 {
205 int rval;
206
207 do
208 {
209 rval= fcntl(ptr->fd, F_SETFL, flags | O_NONBLOCK);
210 }
211 while (rval == -1 && (errno == EINTR || errno == EAGAIN));
212
213 unlikely (rval == -1)
214 {
215 return memcached_set_errno(*ptr, errno, NULL);
216 }
217 }
218 #endif
219 return MEMCACHED_SUCCESS;
220 }
221
222 static memcached_return_t set_socket_options(memcached_server_st *ptr)
223 {
224 WATCHPOINT_ASSERT(ptr->fd != -1);
225
226 if (ptr->type == MEMCACHED_CONNECTION_UDP)
227 return MEMCACHED_SUCCESS;
228
229 #ifdef HAVE_SNDTIMEO
230 if (ptr->root->snd_timeout)
231 {
232 int error;
233 struct timeval waittime;
234
235 waittime.tv_sec= 0;
236 waittime.tv_usec= ptr->root->snd_timeout;
237
238 error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDTIMEO,
239 &waittime, (socklen_t)sizeof(struct timeval));
240 WATCHPOINT_ASSERT(error == 0);
241 if (error)
242 return MEMCACHED_FAILURE;
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 if (error)
259 return MEMCACHED_FAILURE;
260 }
261 #endif
262
263
264 #if defined(__MACH__) && defined(__APPLE__) || defined(__FreeBSD__)
265 {
266 int set = 1;
267 int error= setsockopt(ptr->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
268
269 // This is not considered a fatal error
270 if (error == -1)
271 {
272 WATCHPOINT_ERRNO(get_socket_errno());
273 perror("setsockopt(SO_NOSIGPIPE)");
274 }
275 }
276 #endif
277
278 if (ptr->root->flags.no_block)
279 {
280 int error;
281 struct linger linger;
282
283 linger.l_onoff= 1;
284 linger.l_linger= 0; /* By default on close() just drop the socket */
285 error= setsockopt(ptr->fd, SOL_SOCKET, SO_LINGER,
286 &linger, (socklen_t)sizeof(struct linger));
287 WATCHPOINT_ASSERT(error == 0);
288 if (error)
289 return MEMCACHED_FAILURE;
290 }
291
292 if (ptr->root->flags.tcp_nodelay)
293 {
294 int flag= 1;
295 int error;
296
297 error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_NODELAY,
298 &flag, (socklen_t)sizeof(int));
299 WATCHPOINT_ASSERT(error == 0);
300 if (error)
301 return MEMCACHED_FAILURE;
302 }
303
304 if (ptr->root->flags.tcp_keepalive)
305 {
306 int flag= 1;
307 int error;
308
309 error= setsockopt(ptr->fd, SOL_SOCKET, SO_KEEPALIVE,
310 &flag, (socklen_t)sizeof(int));
311 WATCHPOINT_ASSERT(error == 0);
312 if (error)
313 return MEMCACHED_FAILURE;
314 }
315
316 #ifdef TCP_KEEPIDLE
317 if (ptr->root->tcp_keepidle > 0)
318 {
319 int error;
320
321 error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_KEEPIDLE,
322 &ptr->root->tcp_keepidle, (socklen_t)sizeof(int));
323 WATCHPOINT_ASSERT(error == 0);
324 if (error)
325 return MEMCACHED_FAILURE;
326 }
327 #endif
328
329 if (ptr->root->send_size > 0)
330 {
331 int error;
332
333 error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF,
334 &ptr->root->send_size, (socklen_t)sizeof(int));
335 WATCHPOINT_ASSERT(error == 0);
336 if (error)
337 return MEMCACHED_FAILURE;
338 }
339
340 if (ptr->root->recv_size > 0)
341 {
342 int error;
343
344 error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVBUF,
345 &ptr->root->recv_size, (socklen_t)sizeof(int));
346 WATCHPOINT_ASSERT(error == 0);
347 if (error)
348 return MEMCACHED_FAILURE;
349 }
350
351
352 /* libmemcached will always use nonblocking IO to avoid write deadlocks */
353 return set_socket_nonblocking(ptr);
354 }
355
356 static memcached_return_t unix_socket_connect(memcached_server_st *ptr)
357 {
358 #ifndef WIN32
359 WATCHPOINT_ASSERT(ptr->fd == -1);
360
361 if ((ptr->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
362 {
363 ptr->cached_errno= errno;
364 return MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE;
365 }
366
367 struct sockaddr_un servAddr;
368
369 memset(&servAddr, 0, sizeof (struct sockaddr_un));
370 servAddr.sun_family= AF_UNIX;
371 strncpy(servAddr.sun_path, ptr->hostname, sizeof(servAddr.sun_path)); /* Copy filename */
372
373 test_connect:
374 if (connect(ptr->fd,
375 (struct sockaddr *)&servAddr,
376 sizeof(servAddr)) < 0)
377 {
378 switch (errno)
379 {
380 case EINPROGRESS:
381 case EALREADY:
382 case EINTR:
383 goto test_connect;
384 case EISCONN: /* We were spinning waiting on connect */
385 break;
386 default:
387 WATCHPOINT_ERRNO(errno);
388 ptr->cached_errno= errno;
389 return MEMCACHED_ERRNO;
390 }
391 }
392
393 WATCHPOINT_ASSERT(ptr->fd != -1);
394
395 return MEMCACHED_SUCCESS;
396 #else
397 (void)ptr;
398 return MEMCACHED_NOT_SUPPORTED;
399 #endif
400 }
401
402 static memcached_return_t network_connect(memcached_server_st *ptr)
403 {
404 bool timeout_error_occured= false;
405
406 WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
407 WATCHPOINT_ASSERT(ptr->cursor_active == 0);
408
409 if (! ptr->address_info)
410 {
411 memcached_return_t rc= set_hostinfo(ptr);
412 if (rc != MEMCACHED_SUCCESS)
413 return rc;
414 }
415
416 /* Create the socket */
417 while (ptr->address_info_next && ptr->fd == INVALID_SOCKET)
418 {
419 /* Memcache server does not support IPV6 in udp mode, so skip if not ipv4 */
420 if (ptr->type == MEMCACHED_CONNECTION_UDP && ptr->address_info_next->ai_family != AF_INET)
421 {
422 ptr->address_info_next= ptr->address_info_next->ai_next;
423 continue;
424 }
425
426 if ((ptr->fd= socket(ptr->address_info_next->ai_family,
427 ptr->address_info_next->ai_socktype,
428 ptr->address_info_next->ai_protocol)) < 0)
429 {
430 ptr->cached_errno= get_socket_errno();
431 WATCHPOINT_ERRNO(get_socket_errno());
432 return MEMCACHED_CONNECTION_SOCKET_CREATE_FAILURE;
433 }
434
435 (void)set_socket_options(ptr);
436
437 /* connect to server */
438 if ((connect(ptr->fd, ptr->address_info_next->ai_addr, ptr->address_info_next->ai_addrlen) != SOCKET_ERROR))
439 {
440 break; // Success
441 }
442
443 /* An error occurred */
444 ptr->cached_errno= get_socket_errno();
445 switch (ptr->cached_errno)
446 {
447 case EWOULDBLOCK:
448 case EINPROGRESS: // nonblocking mode - first return
449 case EALREADY: // nonblocking mode - subsequent returns
450 {
451 memcached_return_t rc;
452 rc= connect_poll(ptr);
453
454 if (rc == MEMCACHED_TIMEOUT)
455 timeout_error_occured= true;
456
457 if (rc == MEMCACHED_SUCCESS)
458 break;
459 }
460
461 case EISCONN: // we are connected :-)
462 break;
463
464 case EINTR: // Special case, we retry ai_addr
465 (void)closesocket(ptr->fd);
466 ptr->fd= INVALID_SOCKET;
467 continue;
468
469 default:
470 (void)closesocket(ptr->fd);
471 ptr->fd= INVALID_SOCKET;
472 ptr->address_info_next= ptr->address_info_next->ai_next;
473 break;
474 }
475 }
476
477 if (ptr->fd == INVALID_SOCKET)
478 {
479 WATCHPOINT_STRING("Never got a good file descriptor");
480
481 /* Failed to connect. schedule next retry */
482 if (ptr->root->retry_timeout)
483 {
484 struct timeval next_time;
485
486 if (gettimeofday(&next_time, NULL) == 0)
487 ptr->next_retry= next_time.tv_sec + ptr->root->retry_timeout;
488 }
489
490 if (timeout_error_occured)
491 return MEMCACHED_TIMEOUT;
492
493 return MEMCACHED_ERRNO; /* The last error should be from connect() */
494 }
495
496 return MEMCACHED_SUCCESS; /* The last error should be from connect() */
497 }
498
499 void set_last_disconnected_host(memcached_server_write_instance_st ptr)
500 {
501 // const_cast
502 memcached_st *root= (memcached_st *)ptr->root;
503
504 #if 0
505 WATCHPOINT_STRING(ptr->hostname);
506 WATCHPOINT_NUMBER(ptr->port);
507 WATCHPOINT_ERRNO(ptr->cached_errno);
508 #endif
509 if (root->last_disconnected_server)
510 memcached_server_free(root->last_disconnected_server);
511 root->last_disconnected_server= memcached_server_clone(NULL, ptr);
512 }
513
514 memcached_return_t memcached_connect(memcached_server_write_instance_st ptr)
515 {
516 memcached_return_t rc= MEMCACHED_NO_SERVERS;
517
518 if (ptr->fd != INVALID_SOCKET)
519 return MEMCACHED_SUCCESS;
520
521 LIBMEMCACHED_MEMCACHED_CONNECT_START();
522
523 /* both retry_timeout and server_failure_limit must be set in order to delay retrying a server on error. */
524 WATCHPOINT_ASSERT(ptr->root);
525 if (ptr->root->retry_timeout && ptr->next_retry)
526 {
527 struct timeval curr_time;
528
529 gettimeofday(&curr_time, NULL);
530
531 // We should optimize this to remove the allocation if the server was
532 // the last server to die
533 if (ptr->next_retry > curr_time.tv_sec)
534 {
535 set_last_disconnected_host(ptr);
536
537 return MEMCACHED_SERVER_MARKED_DEAD;
538 }
539 }
540
541 // If we are over the counter failure, we just fail. Reject host only
542 // works if you have a set number of failures.
543 if (ptr->root->server_failure_limit && ptr->server_failure_counter >= ptr->root->server_failure_limit)
544 {
545 set_last_disconnected_host(ptr);
546
547 // @todo fix this by fixing behavior to no longer make use of
548 // memcached_st
549 if (_is_auto_eject_host(ptr->root))
550 {
551 run_distribution((memcached_st *)ptr->root);
552 }
553
554 return MEMCACHED_SERVER_MARKED_DEAD;
555 }
556
557 /* We need to clean up the multi startup piece */
558 switch (ptr->type)
559 {
560 case MEMCACHED_CONNECTION_UNKNOWN:
561 WATCHPOINT_ASSERT(0);
562 rc= MEMCACHED_NOT_SUPPORTED;
563 break;
564 case MEMCACHED_CONNECTION_UDP:
565 case MEMCACHED_CONNECTION_TCP:
566 rc= network_connect(ptr);
567 #ifdef LIBMEMCACHED_WITH_SASL_SUPPORT
568 if (ptr->fd != INVALID_SOCKET && ptr->root->sasl.callbacks)
569 {
570 rc= memcached_sasl_authenticate_connection(ptr);
571 if (rc != MEMCACHED_SUCCESS)
572 {
573 (void)closesocket(ptr->fd);
574 ptr->fd= INVALID_SOCKET;
575 }
576 }
577 #endif
578 break;
579 case MEMCACHED_CONNECTION_UNIX_SOCKET:
580 rc= unix_socket_connect(ptr);
581 break;
582 case MEMCACHED_CONNECTION_MAX:
583 default:
584 WATCHPOINT_ASSERT(0);
585 }
586
587 if (rc == MEMCACHED_SUCCESS)
588 {
589 ptr->server_failure_counter= 0;
590 ptr->next_retry= 0;
591 }
592 else
593 {
594 ptr->server_failure_counter++;
595
596 set_last_disconnected_host(ptr);
597 }
598
599 LIBMEMCACHED_MEMCACHED_CONNECT_END();
600
601 return rc;
602 }