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