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