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