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
  78         return memcached_set_errno(*ptr, err, MEMCACHED_AT);
  79       }
  80     case 0:
  81       {
  82         return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
  83       }
  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         ptr->state= MEMCACHED_SERVER_STATE_NEW;
 118
 119         return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
 120       }
 121     }
 122   }
 123
 124   // This should only be possible from ERESTART or EINTR;
 125   return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
 126 }
 127
 128 static memcached_return_t set_hostinfo(memcached_server_st *server)
 129 {
 130   assert(! server->address_info); // We cover the case where a programming mistake has been made.
 131   if (server->address_info)
 132   {
 133     freeaddrinfo(server->address_info);
 134     server->address_info= NULL;
 135     server->address_info_next= NULL;
 136   }
 137
 138   char str_port[NI_MAXSERV];
 139   int length= snprintf(str_port, NI_MAXSERV, "%u", (uint32_t)server->port);
 140   if (length >= NI_MAXSERV || length < 0)
 141     return MEMCACHED_FAILURE;
 142
 143   struct addrinfo hints;
 144   memset(&hints, 0, sizeof(struct addrinfo));
 145
 146 #if 0
 147   hints.ai_family= AF_INET;
 148 #endif
 149   if (server->type == MEMCACHED_CONNECTION_UDP)
 150   {
 151     hints.ai_protocol= IPPROTO_UDP;
 152     hints.ai_socktype= SOCK_DGRAM;
 153   }
 154   else
 155   {
 156     hints.ai_socktype= SOCK_STREAM;
 157     hints.ai_protocol= IPPROTO_TCP;
 158   }
 159
 160   int errcode;
 161   switch(errcode= getaddrinfo(server->hostname, str_port, &hints, &server->address_info))
 162   {
 163   case 0:
 164     break;
 165
 166   case EAI_AGAIN:
 167     return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
 168
 169   case EAI_SYSTEM:
 170       return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_SYSTEM)"));
 171
 172   case EAI_BADFLAGS:
 173     return memcached_set_error(*server, MEMCACHED_INVALID_ARGUMENTS, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_BADFLAGS)"));
 174
 175   case EAI_MEMORY:
 176     return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
 177
 178   default:
 179     {
 180       WATCHPOINT_STRING(server->hostname);
 181       WATCHPOINT_STRING(gai_strerror(e));
 182       return memcached_set_error(*server, MEMCACHED_HOST_LOOKUP_FAILURE, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
 183     }
 184   }
 185   server->address_info_next= server->address_info;
 186   server->state= MEMCACHED_SERVER_STATE_ADDRINFO;
 187
 188   return MEMCACHED_SUCCESS;
 189 }
 190
 191 static inline void set_socket_nonblocking(memcached_server_st *ptr)
 192 {
 193 #ifdef WIN32
 194   u_long arg = 1;
 195   if (ioctlsocket(ptr->fd, FIONBIO, &arg) == SOCKET_ERROR)
 196   {
 197     memcached_set_errno(*ptr, get_socket_errno(), NULL);
 198   }
 199 #else
 200   int flags;
 201
 202   do
 203   {
 204     flags= fcntl(ptr->fd, F_GETFL, 0);
 205   } while (flags == -1 && (errno == EINTR || errno == EAGAIN));
 206
 207   if (flags == -1)
 208   {
 209     memcached_set_errno(*ptr, errno, NULL);
 210   }
 211   else if ((flags & O_NONBLOCK) == 0)
 212   {
 213     int rval;
 214
 215     do
 216     {
 217       rval= fcntl(ptr->fd, F_SETFL, flags | O_NONBLOCK);
 218     } while (rval == -1 && (errno == EINTR || errno == EAGAIN));
 219
 220     unlikely (rval == -1)
 221     {
 222       memcached_set_errno(*ptr, errno, NULL);
 223     }
 224   }
 225 #endif
 226 }
 227
 228 static void set_socket_options(memcached_server_st *ptr)
 229 {
 230   WATCHPOINT_ASSERT(ptr->fd != -1);
 231
 232   if (ptr->type == MEMCACHED_CONNECTION_UDP)
 233     return;
 234
 235 #ifdef HAVE_SNDTIMEO
 236   if (ptr->root->snd_timeout)
 237   {
 238     int error;
 239     struct timeval waittime;
 240
 241     waittime.tv_sec= 0;
 242     waittime.tv_usec= ptr->root->snd_timeout;
 243
 244     error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDTIMEO,
 245                       &waittime, (socklen_t)sizeof(struct timeval));
 246     WATCHPOINT_ASSERT(error == 0);
 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   }
 263 #endif
 264
 265
 266 #if defined(__MACH__) && defined(__APPLE__) || defined(__FreeBSD__)
 267   {
 268     int set= 1;
 269     int error= setsockopt(ptr->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 270
 271     // This is not considered a fatal error
 272     if (error == -1)
 273     {
 274       WATCHPOINT_ERRNO(get_socket_errno());
 275       perror("setsockopt(SO_NOSIGPIPE)");
 276     }
 277   }
 278 #endif
 279
 280   if (ptr->root->flags.no_block)
 281   {
 282     int error;
 283     struct linger linger;
 284
 285     linger.l_onoff= 1;
 286     linger.l_linger= 0; /* By default on close() just drop the socket */
 287     error= setsockopt(ptr->fd, SOL_SOCKET, SO_LINGER,
 288                       &linger, (socklen_t)sizeof(struct linger));
 289     WATCHPOINT_ASSERT(error == 0);
 290   }
 291
 292   if (ptr->root->flags.tcp_nodelay)
 293   {
 294     int flag= 1;
 295     int error;
 296
 297     error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_NODELAY,
 298                       &flag, (socklen_t)sizeof(int));
 299     WATCHPOINT_ASSERT(error == 0);
 300   }
 301
 302   if (ptr->root->flags.tcp_keepalive)
 303   {
 304     int flag= 1;
 305     int error;
 306
 307     error= setsockopt(ptr->fd, SOL_SOCKET, SO_KEEPALIVE,
 308                       &flag, (socklen_t)sizeof(int));
 309     WATCHPOINT_ASSERT(error == 0);
 310   }
 311
 312 #ifdef TCP_KEEPIDLE
 313   if (ptr->root->tcp_keepidle > 0)
 314   {
 315     int error;
 316
 317     error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_KEEPIDLE,
 318                       &ptr->root->tcp_keepidle, (socklen_t)sizeof(int));
 319     WATCHPOINT_ASSERT(error == 0);
 320   }
 321 #endif
 322
 323   if (ptr->root->send_size > 0)
 324   {
 325     int error;
 326
 327     error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF,
 328                       &ptr->root->send_size, (socklen_t)sizeof(int));
 329     WATCHPOINT_ASSERT(error == 0);
 330   }
 331
 332   if (ptr->root->recv_size > 0)
 333   {
 334     int error;
 335
 336     error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVBUF,
 337                       &ptr->root->recv_size, (socklen_t)sizeof(int));
 338     WATCHPOINT_ASSERT(error == 0);
 339   }
 340
 341
 342   /* libmemcached will always use nonblocking IO to avoid write deadlocks */
 343   set_socket_nonblocking(ptr);
 344 }
 345
 346 static memcached_return_t unix_socket_connect(memcached_server_st *ptr)
 347 {
 348 #ifndef WIN32
 349   WATCHPOINT_ASSERT(ptr->fd == -1);
 350
 351   if ((ptr->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
 352   {
 353     return memcached_set_errno(*ptr, errno, NULL);
 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         return memcached_set_errno(*ptr, errno, MEMCACHED_AT);
 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       (void)closesocket(ptr->fd);
 486       ptr->fd= INVALID_SOCKET;
 487       ptr->state= MEMCACHED_SERVER_STATE_NEW;
 488       continue;
 489
 490     default:
 491       break;
 492     }
 493
 494     (void)closesocket(ptr->fd);
 495     ptr->fd= INVALID_SOCKET;
 496     ptr->address_info_next= ptr->address_info_next->ai_next;
 497   }
 498
 499   WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
 500
 501   if (timeout_error_occured)
 502   {
 503     if (ptr->fd != INVALID_SOCKET)
 504     {
 505       (void)closesocket(ptr->fd);
 506       ptr->fd= INVALID_SOCKET;
 507     }
 508   }
 509
 510   WATCHPOINT_STRING("Never got a good file descriptor");
 511   /* Failed to connect. schedule next retry */
 512   if (ptr->root->retry_timeout)
 513   {
 514     struct timeval next_time;
 515
 516     if (gettimeofday(&next_time, NULL) == 0)
 517       ptr->next_retry= next_time.tv_sec + ptr->root->retry_timeout;
 518   }
 519
 520   if (timeout_error_occured)
 521     return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
 522
 523   return memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); /* The last error should be from connect() */
 524 }
 525
 526 void set_last_disconnected_host(memcached_server_write_instance_st self)
 527 {
 528   // const_cast
 529   memcached_st *root= (memcached_st *)self->root;
 530
 531 #if 0
 532   WATCHPOINT_STRING(self->hostname);
 533   WATCHPOINT_NUMBER(self->port);
 534   WATCHPOINT_ERRNO(self->cached_errno);
 535 #endif
 536   memcached_server_free(root->last_disconnected_server);
 537   root->last_disconnected_server= memcached_server_clone(NULL, self);
 538 }
 539
 540 memcached_return_t memcached_connect(memcached_server_write_instance_st ptr)
 541 {
 542   memcached_return_t rc= MEMCACHED_NO_SERVERS;
 543
 544   if (ptr->fd != INVALID_SOCKET)
 545     return MEMCACHED_SUCCESS;
 546
 547   LIBMEMCACHED_MEMCACHED_CONNECT_START();
 548
 549   /* both retry_timeout and server_failure_limit must be set in order to delay retrying a server on error. */
 550   WATCHPOINT_ASSERT(ptr->root);
 551   if (ptr->root->retry_timeout && ptr->next_retry)
 552   {
 553     struct timeval curr_time;
 554
 555     gettimeofday(&curr_time, NULL);
 556
 557     // We should optimize this to remove the allocation if the server was
 558     // the last server to die
 559     if (ptr->next_retry > curr_time.tv_sec)
 560     {
 561       set_last_disconnected_host(ptr);
 562
 563       return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
 564     }
 565   }
 566
 567   // If we are over the counter failure, we just fail. Reject host only
 568   // works if you have a set number of failures.
 569   if (ptr->root->server_failure_limit && ptr->server_failure_counter >= ptr->root->server_failure_limit)
 570   {
 571     set_last_disconnected_host(ptr);
 572
 573     // @todo fix this by fixing behavior to no longer make use of
 574     // memcached_st
 575     if (_is_auto_eject_host(ptr->root))
 576     {
 577       run_distribution((memcached_st *)ptr->root);
 578     }
 579
 580     return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
 581   }
 582
 583   /* We need to clean up the multi startup piece */
 584   switch (ptr->type)
 585   {
 586   case MEMCACHED_CONNECTION_UNKNOWN:
 587     WATCHPOINT_ASSERT(0);
 588     rc= MEMCACHED_NOT_SUPPORTED;
 589     break;
 590
 591   case MEMCACHED_CONNECTION_UDP:
 592   case MEMCACHED_CONNECTION_TCP:
 593     rc= network_connect(ptr);
 594 #ifdef LIBMEMCACHED_WITH_SASL_SUPPORT
 595     if (ptr->fd != INVALID_SOCKET && ptr->root->sasl.callbacks)
 596     {
 597       rc= memcached_sasl_authenticate_connection(ptr);
 598       if (memcached_failed(rc))
 599       {
 600         (void)closesocket(ptr->fd);
 601         ptr->fd= INVALID_SOCKET;
 602       }
 603     }
 604 #endif
 605     break;
 606
 607   case MEMCACHED_CONNECTION_UNIX_SOCKET:
 608     rc= unix_socket_connect(ptr);
 609     break;
 610
 611   case MEMCACHED_CONNECTION_MAX:
 612   default:
 613     WATCHPOINT_ASSERT(0);
 614   }
 615
 616   if (memcached_success(rc))
 617   {
 618     ptr->server_failure_counter= 0;
 619     ptr->next_retry= 0;
 620   }
 621   else
 622   {
 623     memcached_set_error(*ptr, rc, MEMCACHED_AT);
 624     ptr->server_failure_counter++;
 625     set_last_disconnected_host(ptr);
 626   }
 627
 628   LIBMEMCACHED_MEMCACHED_CONNECT_END();
 629
 630   return rc;
 631 }