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