git.m6w6.name Git - m6w6/libmemcached/blob - libmemcached/connect.c

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