git.m6w6.name Git - m6w6/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       switch (get_socket_errno())
  82       {
  83 #ifdef TARGET_OS_LINUX
  84       case ERESTART:
  85 #endif
  86       case EINTR:
  87         continue;
  88
  89       case EFAULT:
  90       case ENOMEM:
  91         return memcached_set_error(*ptr, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
  92
  93       case EINVAL:
  94         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"));
  95
  96       default: // This should not happen
  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           memcached_set_errno(*ptr, (err == 0) ? get_socket_errno() : err, MEMCACHED_AT);
 103         }
 104         else
 105         {
 106           memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
 107         }
 108
 109         assert_msg(ptr->fd != INVALID_SOCKET, "poll() was passed an invalid file descriptor");
 110         (void)closesocket(ptr->fd);
 111         ptr->fd= INVALID_SOCKET;
 112         ptr->state= MEMCACHED_SERVER_STATE_NEW;
 113
 114         return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
 115       }
 116     }
 117   }
 118
 119   // This should only be possible from ERESTART or EINTR;
 120   return memcached_set_errno(*ptr, get_socket_errno(), MEMCACHED_AT);
 121 }
 122
 123 static memcached_return_t set_hostinfo(memcached_server_st *server)
 124 {
 125   if (server->address_info)
 126   {
 127     freeaddrinfo(server->address_info);
 128     server->address_info= NULL;
 129     server->address_info_next= NULL;
 130   }
 131
 132   char str_port[NI_MAXSERV];
 133   int length= snprintf(str_port, NI_MAXSERV, "%u", (uint32_t)server->port);
 134   if (length >= NI_MAXSERV or length < 0)
 135   {
 136     return MEMCACHED_FAILURE;
 137   }
 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_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
 173
 174   default:
 175     {
 176       return memcached_set_error(*server, MEMCACHED_HOST_LOOKUP_FAILURE, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
 177     }
 178   }
 179   server->address_info_next= server->address_info;
 180   server->state= MEMCACHED_SERVER_STATE_ADDRINFO;
 181
 182   return MEMCACHED_SUCCESS;
 183 }
 184
 185 static inline void set_socket_nonblocking(memcached_server_st *ptr)
 186 {
 187 #ifdef WIN32
 188   u_long arg = 1;
 189   if (ioctlsocket(ptr->fd, FIONBIO, &arg) == SOCKET_ERROR)
 190   {
 191     memcached_set_errno(*ptr, get_socket_errno(), NULL);
 192   }
 193 #else
 194   int flags;
 195
 196   do
 197   {
 198     flags= fcntl(ptr->fd, F_GETFL, 0);
 199   } while (flags == -1 && (errno == EINTR || errno == EAGAIN));
 200
 201   if (flags == -1)
 202   {
 203     memcached_set_errno(*ptr, errno, NULL);
 204   }
 205   else if ((flags & O_NONBLOCK) == 0)
 206   {
 207     int rval;
 208
 209     do
 210     {
 211       rval= fcntl(ptr->fd, F_SETFL, flags | O_NONBLOCK);
 212     } while (rval == -1 && (errno == EINTR || errno == EAGAIN));
 213
 214     unlikely (rval == -1)
 215     {
 216       memcached_set_errno(*ptr, errno, NULL);
 217     }
 218   }
 219 #endif
 220 }
 221
 222 static void set_socket_options(memcached_server_st *ptr)
 223 {
 224   assert_msg(ptr->fd != -1, "invalid socket was passed to set_socket_options()");
 225
 226   if (ptr->type == MEMCACHED_CONNECTION_UDP)
 227   {
 228     return;
 229   }
 230
 231 #ifdef HAVE_SNDTIMEO
 232   if (ptr->root->snd_timeout)
 233   {
 234     int error;
 235     struct timeval waittime;
 236
 237     waittime.tv_sec= 0;
 238     waittime.tv_usec= ptr->root->snd_timeout;
 239
 240     error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDTIMEO,
 241                       &waittime, (socklen_t)sizeof(struct timeval));
 242     WATCHPOINT_ASSERT(error == 0);
 243   }
 244 #endif
 245
 246 #ifdef HAVE_RCVTIMEO
 247   if (ptr->root->rcv_timeout)
 248   {
 249     int error;
 250     struct timeval waittime;
 251
 252     waittime.tv_sec= 0;
 253     waittime.tv_usec= ptr->root->rcv_timeout;
 254
 255     error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVTIMEO,
 256                       &waittime, (socklen_t)sizeof(struct timeval));
 257     WATCHPOINT_ASSERT(error == 0);
 258   }
 259 #endif
 260
 261
 262 #if defined(__MACH__) && defined(__APPLE__) || defined(__FreeBSD__)
 263   {
 264     int set= 1;
 265     int error= setsockopt(ptr->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 266
 267     // This is not considered a fatal error
 268     if (error == -1)
 269     {
 270       WATCHPOINT_ERRNO(get_socket_errno());
 271       perror("setsockopt(SO_NOSIGPIPE)");
 272     }
 273   }
 274 #endif
 275
 276   if (ptr->root->flags.no_block)
 277   {
 278     int error;
 279     struct linger linger;
 280
 281     linger.l_onoff= 1;
 282     linger.l_linger= 0; /* By default on close() just drop the socket */
 283     error= setsockopt(ptr->fd, SOL_SOCKET, SO_LINGER,
 284                       &linger, (socklen_t)sizeof(struct linger));
 285     WATCHPOINT_ASSERT(error == 0);
 286   }
 287
 288   if (ptr->root->flags.tcp_nodelay)
 289   {
 290     int flag= 1;
 291     int error;
 292
 293     error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_NODELAY,
 294                       &flag, (socklen_t)sizeof(int));
 295     WATCHPOINT_ASSERT(error == 0);
 296   }
 297
 298   if (ptr->root->flags.tcp_keepalive)
 299   {
 300     int flag= 1;
 301     int error;
 302
 303     error= setsockopt(ptr->fd, SOL_SOCKET, SO_KEEPALIVE,
 304                       &flag, (socklen_t)sizeof(int));
 305     WATCHPOINT_ASSERT(error == 0);
 306   }
 307
 308 #ifdef TCP_KEEPIDLE
 309   if (ptr->root->tcp_keepidle > 0)
 310   {
 311     int error;
 312
 313     error= setsockopt(ptr->fd, IPPROTO_TCP, TCP_KEEPIDLE,
 314                       &ptr->root->tcp_keepidle, (socklen_t)sizeof(int));
 315     WATCHPOINT_ASSERT(error == 0);
 316   }
 317 #endif
 318
 319   if (ptr->root->send_size > 0)
 320   {
 321     int error;
 322
 323     error= setsockopt(ptr->fd, SOL_SOCKET, SO_SNDBUF,
 324                       &ptr->root->send_size, (socklen_t)sizeof(int));
 325     WATCHPOINT_ASSERT(error == 0);
 326   }
 327
 328   if (ptr->root->recv_size > 0)
 329   {
 330     int error;
 331
 332     error= setsockopt(ptr->fd, SOL_SOCKET, SO_RCVBUF,
 333                       &ptr->root->recv_size, (socklen_t)sizeof(int));
 334     WATCHPOINT_ASSERT(error == 0);
 335   }
 336
 337
 338   /* libmemcached will always use nonblocking IO to avoid write deadlocks */
 339   set_socket_nonblocking(ptr);
 340 }
 341
 342 static memcached_return_t unix_socket_connect(memcached_server_st *ptr)
 343 {
 344 #ifndef WIN32
 345   WATCHPOINT_ASSERT(ptr->fd == -1);
 346
 347   if ((ptr->fd= socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
 348   {
 349     memcached_set_errno(*ptr, errno, NULL);
 350     return MEMCACHED_CONNECTION_FAILURE;
 351   }
 352
 353   struct sockaddr_un servAddr;
 354
 355   memset(&servAddr, 0, sizeof (struct sockaddr_un));
 356   servAddr.sun_family= AF_UNIX;
 357   strncpy(servAddr.sun_path, ptr->hostname, sizeof(servAddr.sun_path)); /* Copy filename */
 358
 359   do {
 360     if (connect(ptr->fd, (struct sockaddr *)&servAddr, sizeof(servAddr)) < 0)
 361     {
 362       switch (errno)
 363       {
 364       case EINPROGRESS:
 365       case EALREADY:
 366       case EINTR:
 367         continue;
 368
 369       case EISCONN: /* We were spinning waiting on connect */
 370         {
 371           WATCHPOINT_ASSERT(0); // Programmer error
 372           break;
 373         }
 374
 375       default:
 376         WATCHPOINT_ERRNO(errno);
 377         memcached_set_errno(*ptr, errno, MEMCACHED_AT);
 378         return MEMCACHED_CONNECTION_FAILURE;
 379       }
 380     }
 381   } while (0);
 382   ptr->state= MEMCACHED_SERVER_STATE_CONNECTED;
 383
 384   WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
 385
 386   return MEMCACHED_SUCCESS;
 387 #else
 388   (void)ptr;
 389   return MEMCACHED_NOT_SUPPORTED;
 390 #endif
 391 }
 392
 393 static memcached_return_t network_connect(memcached_server_st *ptr)
 394 {
 395   bool timeout_error_occured= false;
 396
 397   WATCHPOINT_ASSERT(ptr->fd == INVALID_SOCKET);
 398   WATCHPOINT_ASSERT(ptr->cursor_active == 0);
 399
 400   if (not ptr->address_info)
 401   {
 402     WATCHPOINT_ASSERT(ptr->state == MEMCACHED_SERVER_STATE_NEW);
 403     memcached_return_t rc;
 404     uint32_t counter= 5;
 405     while (--counter)
 406     {
 407       if ((rc= set_hostinfo(ptr)) != MEMCACHED_TIMEOUT)
 408       {
 409         break;
 410       }
 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     {
 519       ptr->next_retry= next_time.tv_sec + ptr->root->retry_timeout;
 520     }
 521   }
 522
 523   if (memcached_has_current_error(*ptr))
 524   {
 525     return memcached_server_error_return(ptr);
 526   }
 527
 528   if (timeout_error_occured and ptr->state < MEMCACHED_SERVER_STATE_IN_PROGRESS)
 529   {
 530     return memcached_set_error(*ptr, MEMCACHED_TIMEOUT, MEMCACHED_AT);
 531   }
 532
 533   return memcached_set_error(*ptr, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); /* The last error should be from connect() */
 534 }
 535
 536 void set_last_disconnected_host(memcached_server_write_instance_st self)
 537 {
 538   // const_cast
 539   memcached_st *root= (memcached_st *)self->root;
 540
 541   memcached_server_free(root->last_disconnected_server);
 542   root->last_disconnected_server= memcached_server_clone(NULL, self);
 543 }
 544
 545 memcached_return_t memcached_connect(memcached_server_write_instance_st ptr)
 546 {
 547   memcached_return_t rc= MEMCACHED_NO_SERVERS;
 548
 549   if (ptr->fd != INVALID_SOCKET)
 550   {
 551     return MEMCACHED_SUCCESS;
 552   }
 553
 554   LIBMEMCACHED_MEMCACHED_CONNECT_START();
 555
 556   /* both retry_timeout and server_failure_limit must be set in order to delay retrying a server on error. */
 557   WATCHPOINT_ASSERT(ptr->root);
 558   if (ptr->root->retry_timeout && ptr->next_retry)
 559   {
 560     struct timeval curr_time;
 561
 562     gettimeofday(&curr_time, NULL);
 563
 564     // We should optimize this to remove the allocation if the server was
 565     // the last server to die
 566     if (ptr->next_retry > curr_time.tv_sec)
 567     {
 568       set_last_disconnected_host(ptr);
 569
 570       return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
 571     }
 572   }
 573
 574   // If we are over the counter failure, we just fail. Reject host only
 575   // works if you have a set number of failures.
 576   if (ptr->root->server_failure_limit && ptr->server_failure_counter >= ptr->root->server_failure_limit)
 577   {
 578     set_last_disconnected_host(ptr);
 579
 580     // @todo fix this by fixing behavior to no longer make use of
 581     // memcached_st
 582     if (_is_auto_eject_host(ptr->root))
 583     {
 584       run_distribution((memcached_st *)ptr->root);
 585     }
 586
 587     return memcached_set_error(*ptr, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
 588   }
 589
 590   /* We need to clean up the multi startup piece */
 591   switch (ptr->type)
 592   {
 593   case MEMCACHED_CONNECTION_UNKNOWN:
 594     WATCHPOINT_ASSERT(0);
 595     rc= MEMCACHED_NOT_SUPPORTED;
 596     break;
 597
 598   case MEMCACHED_CONNECTION_UDP:
 599   case MEMCACHED_CONNECTION_TCP:
 600     rc= network_connect(ptr);
 601 #ifdef LIBMEMCACHED_WITH_SASL_SUPPORT
 602     if (ptr->fd != INVALID_SOCKET and ptr->root->sasl.callbacks)
 603     {
 604       rc= memcached_sasl_authenticate_connection(ptr);
 605       if (memcached_failed(rc) and ptr->fd != INVALID_SOCKET)
 606       {
 607         WATCHPOINT_ASSERT(ptr->fd != INVALID_SOCKET);
 608         (void)closesocket(ptr->fd);
 609         ptr->fd= INVALID_SOCKET;
 610       }
 611     }
 612 #endif
 613     break;
 614
 615   case MEMCACHED_CONNECTION_UNIX_SOCKET:
 616     rc= unix_socket_connect(ptr);
 617     break;
 618
 619   case MEMCACHED_CONNECTION_MAX:
 620   default:
 621     WATCHPOINT_ASSERT(0);
 622   }
 623
 624   if (memcached_success(rc))
 625   {
 626     ptr->server_failure_counter= 0;
 627     ptr->next_retry= 0;
 628   }
 629   else if (memcached_has_current_error(*ptr))
 630   {
 631     ptr->server_failure_counter++;
 632     set_last_disconnected_host(ptr);
 633   }
 634   else
 635   {
 636     memcached_set_error(*ptr, rc, MEMCACHED_AT);
 637     ptr->server_failure_counter++;
 638     set_last_disconnected_host(ptr);
 639   }
 640
 641   LIBMEMCACHED_MEMCACHED_CONNECT_END();
 642
 643   return rc;
 644 }