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
  41 #include <cassert>
  42
  43 #ifndef SOCK_CLOEXEC
  44 #  define SOCK_CLOEXEC 0
  45 #endif
  46
  47 #ifndef SOCK_NONBLOCK
  48 # define SOCK_NONBLOCK 0
  49 #endif
  50
  51 #ifndef FD_CLOEXEC
  52 # define FD_CLOEXEC 0
  53 #endif
  54
  55 #ifndef SO_NOSIGPIPE
  56 # define SO_NOSIGPIPE 0
  57 #endif
  58
  59 #ifndef TCP_NODELAY
  60 # define TCP_NODELAY 0
  61 #endif
  62
  63 #ifndef TCP_KEEPIDLE
  64 # define TCP_KEEPIDLE 0
  65 #endif
  66
  67 static memcached_return_t connect_poll(memcached_instance_st* server, const int connection_error)
  68 {
  69   struct pollfd fds[1];
  70   fds[0].fd= server->fd;
  71   fds[0].events= server->events();
  72   fds[0].revents= 0;
  73
  74   size_t loop_max= 5;
  75
  76   if (server->root->poll_timeout == 0)
  77   {
  78     return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT,
  79                                memcached_literal_param("The time to wait for a connection to be established was set to zero which produces a timeout to every call to poll()."));
  80   }
  81
  82   while (--loop_max) // Should only loop on cases of ERESTART or EINTR
  83   {
  84     int number_of;
  85     if ((number_of= poll(fds, 1, server->root->connect_timeout)) == -1)
  86     {
  87       int local_errno= get_socket_errno(); // We cache in case closesocket() modifies errno
  88       switch (local_errno)
  89       {
  90 #ifdef __linux__
  91       case ERESTART:
  92 #endif
  93       case EINTR:
  94         continue;
  95
  96       case EFAULT:
  97       case ENOMEM:
  98         return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT);
  99
 100       case EINVAL:
 101         return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
 102                                    memcached_literal_param("RLIMIT_NOFILE exceeded, or if OSX the timeout value was invalid"));
 103
 104       default: // This should not happen
 105         break;
 106       }
 107
 108       assert_msg(server->fd != INVALID_SOCKET, "poll() was passed an invalid file descriptor");
 109       server->reset_socket();
 110       server->state= MEMCACHED_SERVER_STATE_NEW;
 111
 112       return memcached_set_errno(*server, local_errno, MEMCACHED_AT);
 113     }
 114
 115     if (number_of == 0)
 116     {
 117       if (connection_error == EINPROGRESS)
 118       {
 119         int err;
 120         socklen_t len= sizeof(err);
 121         if (getsockopt(server->fd, SOL_SOCKET, SO_ERROR, (char*)&err, &len) == -1)
 122         {
 123           return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getsockopt() error'ed while looking for error connect_poll(EINPROGRESS)"));
 124         }
 125
 126         // If Zero, my hero, we just fail to a generic MEMCACHED_TIMEOUT error
 127         if (err != 0)
 128         {
 129           return memcached_set_errno(*server, err, MEMCACHED_AT, memcached_literal_param("getsockopt() found the error from poll() after connect() returned EINPROGRESS."));
 130         }
 131       }
 132
 133       return  memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT, memcached_literal_param("(number_of == 0)"));
 134     }
 135
 136     assert (number_of == 1);
 137
 138     if (fds[0].revents & POLLERR or
 139         fds[0].revents & POLLHUP or
 140         fds[0].revents & POLLNVAL)
 141     {
 142       int err;
 143       socklen_t len= sizeof (err);
 144       if (getsockopt(fds[0].fd, SOL_SOCKET, SO_ERROR, (char*)&err, &len) == -1)
 145       {
 146         return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getsockopt() errored while looking up error state from poll()"));
 147       }
 148
 149       // We check the value to see what happened wth the socket.
 150       if (err == 0) // Should not happen
 151       {
 152         return MEMCACHED_SUCCESS;
 153       }
 154       errno= err;
 155
 156       return memcached_set_errno(*server, err, MEMCACHED_AT, memcached_literal_param("getsockopt() found the error from poll() during connect."));
 157     }
 158     assert(fds[0].revents & POLLOUT);
 159
 160     if (fds[0].revents & POLLOUT and connection_error == EINPROGRESS)
 161     {
 162       int err;
 163       socklen_t len= sizeof(err);
 164       if (getsockopt(server->fd, SOL_SOCKET, SO_ERROR, (char*)&err, &len) == -1)
 165       {
 166         return memcached_set_errno(*server, errno, MEMCACHED_AT);
 167       }
 168
 169       if (err == 0)
 170       {
 171         return MEMCACHED_SUCCESS;
 172       }
 173
 174       return memcached_set_errno(*server, err, MEMCACHED_AT, memcached_literal_param("getsockopt() found the error from poll() after connect() returned EINPROGRESS."));
 175     }
 176
 177     break; // We only have the loop setup for errno types that require restart
 178   }
 179
 180   // This should only be possible from ERESTART or EINTR;
 181   return memcached_set_errno(*server, connection_error, MEMCACHED_AT, memcached_literal_param("connect_poll() was exhausted"));
 182 }
 183
 184 static memcached_return_t set_hostinfo(memcached_instance_st* server)
 185 {
 186   assert(server->type != MEMCACHED_CONNECTION_UNIX_SOCKET);
 187   server->clear_addrinfo();
 188
 189   char str_port[MEMCACHED_NI_MAXSERV]= { 0 };
 190   errno= 0;
 191   int length= snprintf(str_port, MEMCACHED_NI_MAXSERV, "%u", uint32_t(server->port()));
 192   if (length >= MEMCACHED_NI_MAXSERV or length <= 0 or errno != 0)
 193   {
 194     return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT,
 195                                memcached_literal_param("snprintf(NI_MAXSERV)"));
 196   }
 197
 198   struct addrinfo hints;
 199   memset(&hints, 0, sizeof(struct addrinfo));
 200
 201   hints.ai_family= AF_UNSPEC;
 202   if (memcached_is_udp(server->root))
 203   {
 204     hints.ai_protocol= IPPROTO_UDP;
 205     hints.ai_socktype= SOCK_DGRAM;
 206   }
 207   else
 208   {
 209     hints.ai_socktype= SOCK_STREAM;
 210     hints.ai_protocol= IPPROTO_TCP;
 211   }
 212
 213   assert(server->address_info == NULL);
 214   assert(server->address_info_next == NULL);
 215   int errcode;
 216   assert(server->hostname());
 217   switch(errcode= getaddrinfo(server->hostname(), str_port, &hints, &server->address_info))
 218   {
 219   case 0:
 220     server->address_info_next= server->address_info;
 221     server->state= MEMCACHED_SERVER_STATE_ADDRINFO;
 222     break;
 223
 224   case EAI_AGAIN:
 225     return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
 226
 227   case EAI_SYSTEM:
 228     server->clear_addrinfo();
 229     return memcached_set_errno(*server, errno, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_SYSTEM)"));
 230
 231   case EAI_BADFLAGS:
 232     server->clear_addrinfo();
 233     return memcached_set_error(*server, MEMCACHED_INVALID_ARGUMENTS, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_BADFLAGS)"));
 234
 235   case EAI_MEMORY:
 236     server->clear_addrinfo();
 237     return memcached_set_error(*server, MEMCACHED_MEMORY_ALLOCATION_FAILURE, MEMCACHED_AT, memcached_literal_param("getaddrinfo(EAI_MEMORY)"));
 238
 239   default:
 240     {
 241       server->clear_addrinfo();
 242       return memcached_set_error(*server, MEMCACHED_HOST_LOOKUP_FAILURE, MEMCACHED_AT, memcached_string_make_from_cstr(gai_strerror(errcode)));
 243     }
 244   }
 245
 246   return MEMCACHED_SUCCESS;
 247 }
 248
 249 static inline void set_socket_nonblocking(memcached_instance_st* server)
 250 {
 251 #if defined(_WIN32)
 252   u_long arg= 1;
 253   if (ioctlsocket(server->fd, FIONBIO, &arg) == SOCKET_ERROR)
 254   {
 255     memcached_set_errno(*server, get_socket_errno(), NULL);
 256   }
 257 #else
 258   int flags;
 259
 260   if (SOCK_NONBLOCK == 0)
 261   {
 262     do
 263     {
 264       flags= fcntl(server->fd, F_GETFL, 0);
 265     } while (flags == -1 && (errno == EINTR || errno == EAGAIN));
 266
 267     if (flags == -1)
 268     {
 269       memcached_set_errno(*server, errno, NULL);
 270     }
 271     else if ((flags & O_NONBLOCK) == 0)
 272     {
 273       int rval;
 274
 275       do
 276       {
 277         rval= fcntl(server->fd, F_SETFL, flags | O_NONBLOCK);
 278       } while (rval == -1 && (errno == EINTR or errno == EAGAIN));
 279
 280       if (rval == -1)
 281       {
 282         memcached_set_errno(*server, errno, NULL);
 283       }
 284     }
 285   }
 286 #endif
 287 }
 288
 289 static bool set_socket_options(memcached_instance_st* server)
 290 {
 291   assert_msg(server->fd != INVALID_SOCKET, "invalid socket was passed to set_socket_options()");
 292
 293 #ifdef HAVE_FCNTL
 294   // If SOCK_CLOEXEC exists then we don't need to call the following
 295   if (SOCK_CLOEXEC == 0)
 296   {
 297     if (FD_CLOEXEC != 0)
 298     {
 299       int flags;
 300       do
 301       {
 302         flags= fcntl(server->fd, F_GETFD, 0);
 303       } while (flags == -1 and (errno == EINTR or errno == EAGAIN));
 304
 305       if (flags != -1)
 306       {
 307         int rval;
 308         do
 309         {
 310           rval= fcntl (server->fd, F_SETFD, flags | FD_CLOEXEC);
 311         } while (rval == -1 && (errno == EINTR or errno == EAGAIN));
 312         // we currently ignore the case where rval is -1
 313       }
 314     }
 315   }
 316 #endif
 317
 318   if (memcached_is_udp(server->root))
 319   {
 320     return true;
 321   }
 322
 323 #ifdef HAVE_SNDTIMEO
 324   if (server->root->snd_timeout > 0)
 325   {
 326     struct timeval waittime;
 327
 328     waittime.tv_sec= server->root->snd_timeout / 1000000;
 329     waittime.tv_usec= server->root->snd_timeout % 1000000;
 330
 331     int error= setsockopt(server->fd, SOL_SOCKET, SO_SNDTIMEO,
 332                           (char*)&waittime, (socklen_t)sizeof(struct timeval));
 333     (void)error;
 334     assert(error == 0);
 335   }
 336 #endif
 337
 338 #ifdef HAVE_RCVTIMEO
 339   if (server->root->rcv_timeout > 0)
 340   {
 341     struct timeval waittime;
 342
 343     waittime.tv_sec= server->root->rcv_timeout / 1000000;
 344     waittime.tv_usec= server->root->rcv_timeout % 1000000;
 345
 346     int error= setsockopt(server->fd, SOL_SOCKET, SO_RCVTIMEO,
 347                           (char*)&waittime, (socklen_t)sizeof(struct timeval));
 348     (void)(error);
 349     assert(error == 0);
 350   }
 351 #endif
 352
 353
 354 #if defined(_WIN32)
 355 #else
 356 # if defined(SO_NOSIGPIPE)
 357   if (SO_NOSIGPIPE)
 358   {
 359     int set= 1;
 360     int error= setsockopt(server->fd, SOL_SOCKET, SO_NOSIGPIPE, (void *)&set, sizeof(int));
 361
 362     assert(error == 0);
 363
 364     // This is not considered a fatal error
 365     if (error == -1)
 366     {
 367 #if 0
 368       perror("setsockopt(SO_NOSIGPIPE)");
 369 #endif
 370     }
 371   }
 372 # endif // SO_NOSIGPIPE
 373 #endif // _WIN32
 374
 375   if (server->root->flags.no_block)
 376   {
 377     struct linger linger;
 378
 379     linger.l_onoff= 1;
 380     linger.l_linger= 0; /* By default on close() just drop the socket */
 381     int error= setsockopt(server->fd, SOL_SOCKET, SO_LINGER,
 382                           (char*)&linger, (socklen_t)sizeof(struct linger));
 383     (void)(error);
 384     assert(error == 0);
 385   }
 386
 387   if (TCP_NODELAY)
 388   {
 389     if (server->root->flags.tcp_nodelay)
 390     {
 391       int flag= 1;
 392
 393       int error= setsockopt(server->fd, IPPROTO_TCP, TCP_NODELAY,
 394                             (char*)&flag, (socklen_t)sizeof(int));
 395       (void)(error);
 396       assert(error == 0);
 397     }
 398   }
 399
 400   if (server->root->flags.tcp_keepalive)
 401   {
 402     int flag= 1;
 403
 404     int error= setsockopt(server->fd, SOL_SOCKET, SO_KEEPALIVE,
 405                           (char*)&flag, (socklen_t)sizeof(int));
 406     (void)(error);
 407     assert(error == 0);
 408   }
 409
 410   if (TCP_KEEPIDLE)
 411   {
 412     if (server->root->tcp_keepidle > 0)
 413     {
 414       int error= setsockopt(server->fd, IPPROTO_TCP, TCP_KEEPIDLE,
 415                             (char*)&server->root->tcp_keepidle, (socklen_t)sizeof(int));
 416       (void)(error);
 417       assert(error == 0);
 418     }
 419   }
 420
 421   if (server->root->send_size > 0)
 422   {
 423     int error= setsockopt(server->fd, SOL_SOCKET, SO_SNDBUF,
 424                           (char*)&server->root->send_size, (socklen_t)sizeof(int));
 425     (void)(error);
 426     assert(error == 0);
 427   }
 428
 429   if (server->root->recv_size > 0)
 430   {
 431     int error= setsockopt(server->fd, SOL_SOCKET, SO_RCVBUF,
 432                           (char*)&server->root->recv_size, (socklen_t)sizeof(int));
 433     (void)(error);
 434     assert(error == 0);
 435   }
 436
 437   /* libmemcached will always use nonblocking IO to avoid write deadlocks */
 438   set_socket_nonblocking(server);
 439
 440   return true;
 441 }
 442
 443 static memcached_return_t unix_socket_connect(memcached_instance_st* server)
 444 {
 445 #ifndef _WIN32
 446   WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
 447
 448   do {
 449     int type= SOCK_STREAM;
 450     if (SOCK_CLOEXEC != 0)
 451     {
 452       type|= SOCK_CLOEXEC;
 453     }
 454
 455     if (SOCK_NONBLOCK != 0)
 456     {
 457       type|= SOCK_NONBLOCK;
 458     }
 459
 460     if ((server->fd= socket(AF_UNIX, type, 0)) == -1)
 461     {
 462       return memcached_set_errno(*server, errno, NULL);
 463     }
 464
 465     struct sockaddr_un servAddr;
 466
 467     memset(&servAddr, 0, sizeof (struct sockaddr_un));
 468     servAddr.sun_family= AF_UNIX;
 469     if (strlen(server->hostname()) >= sizeof(servAddr.sun_path)) {
 470         return memcached_set_error(*server, MEMCACHED_UNIX_SOCKET_PATH_TOO_BIG, MEMCACHED_AT);
 471     }
 472     strncpy(servAddr.sun_path, server->hostname(), sizeof(servAddr.sun_path)-1); /* Copy filename */
 473
 474     if (connect(server->fd, (struct sockaddr *)&servAddr, sizeof(servAddr)) == -1)
 475     {
 476       switch (errno)
 477       {
 478       case EINPROGRESS:
 479       case EALREADY:
 480         server->events(POLLOUT);
 481         break;
 482
 483       case EINTR:
 484         server->reset_socket();
 485         continue;
 486
 487       case EISCONN: /* We were spinning waiting on connect */
 488         {
 489           assert(0); // Programmer error
 490           server->reset_socket();
 491           continue;
 492         }
 493
 494       default:
 495         WATCHPOINT_ERRNO(errno);
 496         server->reset_socket();
 497         return memcached_set_errno(*server, errno, MEMCACHED_AT);
 498       }
 499     }
 500   } while (0);
 501   server->state= MEMCACHED_SERVER_STATE_CONNECTED;
 502
 503   WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
 504
 505   return MEMCACHED_SUCCESS;
 506 #else
 507   (void)server;
 508   return MEMCACHED_NOT_SUPPORTED;
 509 #endif
 510 }
 511
 512 static memcached_return_t network_connect(memcached_instance_st* server)
 513 {
 514   bool timeout_error_occured= false;
 515
 516   WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
 517   WATCHPOINT_ASSERT(server->cursor_active_ == 0);
 518
 519   /*
 520     We want to check both of these because if address_info_next has been fully tried, we want to do a new lookup to make sure we have picked up on any new DNS information.
 521   */
 522   if (server->address_info == NULL or server->address_info_next == NULL)
 523   {
 524     WATCHPOINT_ASSERT(server->state == MEMCACHED_SERVER_STATE_NEW);
 525     server->address_info_next= NULL;
 526     memcached_return_t rc= set_hostinfo(server);
 527
 528     if (memcached_failed(rc))
 529     {
 530       return rc;
 531     }
 532   }
 533
 534   assert(server->address_info_next);
 535   assert(server->address_info);
 536
 537   /* Create the socket */
 538   while (server->address_info_next and server->fd == INVALID_SOCKET)
 539   {
 540     /* Memcache server does not support IPV6 in udp mode, so skip if not ipv4 */
 541     if (memcached_is_udp(server->root) and server->address_info_next->ai_family != AF_INET)
 542     {
 543       server->address_info_next= server->address_info_next->ai_next;
 544       continue;
 545     }
 546
 547     int type= server->address_info_next->ai_socktype;
 548     if (SOCK_CLOEXEC != 0)
 549     {
 550       type|= SOCK_CLOEXEC;
 551     }
 552
 553     if (SOCK_NONBLOCK != 0)
 554     {
 555       type|= SOCK_NONBLOCK;
 556     }
 557
 558     server->fd= socket(server->address_info_next->ai_family,
 559                        type,
 560                        server->address_info_next->ai_protocol);
 561
 562     if (int(server->fd) == SOCKET_ERROR)
 563     {
 564       return memcached_set_errno(*server, get_socket_errno(), NULL);
 565     }
 566
 567     if (set_socket_options(server) == false)
 568     {
 569       server->reset_socket();
 570       return MEMCACHED_CONNECTION_FAILURE;
 571     }
 572
 573     /* connect to server */
 574     if ((connect(server->fd, server->address_info_next->ai_addr, server->address_info_next->ai_addrlen) != SOCKET_ERROR))
 575     {
 576       server->state= MEMCACHED_SERVER_STATE_CONNECTED;
 577       return MEMCACHED_SUCCESS;
 578     }
 579
 580     /* An error occurred */
 581     int local_error= get_socket_errno();
 582     switch (local_error)
 583     {
 584     case ETIMEDOUT:
 585       timeout_error_occured= true;
 586       break;
 587
 588     case EAGAIN:
 589 #if EWOULDBLOCK != EAGAIN
 590     case EWOULDBLOCK:
 591 #endif
 592     case EINPROGRESS: // nonblocking mode - first return
 593     case EALREADY: // nonblocking mode - subsequent returns
 594       {
 595         server->events(POLLOUT);
 596         server->state= MEMCACHED_SERVER_STATE_IN_PROGRESS;
 597         memcached_return_t rc= connect_poll(server, local_error);
 598
 599         if (memcached_success(rc))
 600         {
 601           server->state= MEMCACHED_SERVER_STATE_CONNECTED;
 602           return MEMCACHED_SUCCESS;
 603         }
 604
 605         // A timeout here is treated as an error, we will not retry
 606         if (rc == MEMCACHED_TIMEOUT)
 607         {
 608           timeout_error_occured= true;
 609         }
 610       }
 611       break;
 612
 613     case EISCONN: // we are connected :-)
 614       WATCHPOINT_ASSERT(0); // This is a programmer's error
 615       break;
 616
 617     case EINTR: // Special case, we retry ai_addr
 618       WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
 619       server->reset_socket();
 620       continue;
 621
 622     case ECONNREFUSED:
 623       // Probably not running service
 624
 625     default:
 626       break;
 627     }
 628
 629     WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
 630     server->reset_socket();
 631     server->address_info_next= server->address_info_next->ai_next;
 632   }
 633
 634   WATCHPOINT_ASSERT(server->fd == INVALID_SOCKET);
 635
 636   if (timeout_error_occured)
 637   {
 638     server->reset_socket();
 639   }
 640
 641   WATCHPOINT_STRING("Never got a good file descriptor");
 642
 643   if (memcached_has_current_error(*server))
 644   {
 645     return memcached_instance_error_return(server);
 646   }
 647
 648   if (timeout_error_occured and server->state < MEMCACHED_SERVER_STATE_IN_PROGRESS)
 649   {
 650     return memcached_set_error(*server, MEMCACHED_TIMEOUT, MEMCACHED_AT,
 651                                memcached_literal_param("if (timeout_error_occured and server->state < MEMCACHED_SERVER_STATE_IN_PROGRESS)"));
 652   }
 653
 654   return memcached_set_error(*server, MEMCACHED_CONNECTION_FAILURE, MEMCACHED_AT); /* The last error should be from connect() */
 655 }
 656
 657
 658 /*
 659   backoff_handling()
 660
 661   Based on time/failure count fail the connect without trying. This prevents waiting in a state where
 662   we get caught spending cycles just waiting.
 663 */
 664 static memcached_return_t backoff_handling(memcached_instance_st* server, bool& in_timeout)
 665 {
 666   struct timeval curr_time;
 667   bool _gettime_success= (gettimeofday(&curr_time, NULL) == 0);
 668
 669   /*
 670     If we hit server_failure_limit then something is completely wrong about the server.
 671
 672     1) If autoeject is enabled we do that.
 673     2) If not? We go into timeout again, there is much else to do :(
 674   */
 675   if (server->server_failure_counter >= server->root->server_failure_limit)
 676   {
 677     /*
 678       We just auto_eject if we hit this point
 679     */
 680     if (_is_auto_eject_host(server->root))
 681     {
 682       set_last_disconnected_host(server);
 683
 684       // Retry dead servers if requested
 685       if (_gettime_success and server->root->dead_timeout > 0)
 686       {
 687         server->next_retry= curr_time.tv_sec +server->root->dead_timeout;
 688
 689         // We only retry dead servers once before assuming failure again
 690         server->server_failure_counter= server->root->server_failure_limit -1;
 691       }
 692
 693       memcached_return_t rc;
 694       if (memcached_failed(rc= run_distribution((memcached_st *)server->root)))
 695       {
 696         return memcached_set_error(*server, rc, MEMCACHED_AT, memcached_literal_param("Backoff handling failed during run_distribution"));
 697       }
 698
 699       return memcached_set_error(*server, MEMCACHED_SERVER_MARKED_DEAD, MEMCACHED_AT);
 700     }
 701
 702     server->state= MEMCACHED_SERVER_STATE_IN_TIMEOUT;
 703
 704     // Sanity check/setting
 705     if (server->next_retry == 0)
 706     {
 707       server->next_retry= 1;
 708     }
 709   }
 710
 711   if (server->state == MEMCACHED_SERVER_STATE_IN_TIMEOUT)
 712   {
 713     /*
 714       If next_retry is less then our current time, then we reset and try everything again.
 715     */
 716     if (_gettime_success and server->next_retry < curr_time.tv_sec)
 717     {
 718       server->state= MEMCACHED_SERVER_STATE_NEW;
 719       server->server_timeout_counter= 0;
 720     }
 721     else
 722     {
 723       return memcached_set_error(*server, MEMCACHED_SERVER_TEMPORARILY_DISABLED, MEMCACHED_AT);
 724     }
 725
 726     in_timeout= true;
 727   }
 728
 729   return MEMCACHED_SUCCESS;
 730 }
 731
 732 static memcached_return_t _memcached_connect(memcached_instance_st* server, const bool set_last_disconnected)
 733 {
 734   assert(server);
 735   if (server->fd != INVALID_SOCKET)
 736   {
 737     return MEMCACHED_SUCCESS;
 738   }
 739
 740   LIBMEMCACHED_MEMCACHED_CONNECT_START();
 741
 742   bool in_timeout= false;
 743   memcached_return_t rc;
 744   if (memcached_failed(rc= backoff_handling(server, in_timeout)))
 745   {
 746     set_last_disconnected_host(server);
 747     return rc;
 748   }
 749
 750   if (LIBMEMCACHED_WITH_SASL_SUPPORT and server->root->sasl.callbacks and memcached_is_udp(server->root))
 751   {
 752     return memcached_set_error(*server, MEMCACHED_INVALID_HOST_PROTOCOL, MEMCACHED_AT, memcached_literal_param("SASL is not supported for UDP connections"));
 753   }
 754
 755   if (server->hostname()[0] == '/')
 756   {
 757     server->type= MEMCACHED_CONNECTION_UNIX_SOCKET;
 758   }
 759
 760   /* We need to clean up the multi startup piece */
 761   switch (server->type)
 762   {
 763   case MEMCACHED_CONNECTION_UDP:
 764   case MEMCACHED_CONNECTION_TCP:
 765     rc= network_connect(server);
 766
 767 #if defined(LIBMEMCACHED_WITH_SASL_SUPPORT)
 768     if (LIBMEMCACHED_WITH_SASL_SUPPORT)
 769     {
 770       if (server->fd != INVALID_SOCKET and server->root->sasl.callbacks)
 771       {
 772         rc= memcached_sasl_authenticate_connection(server);
 773         if (memcached_failed(rc) and server->fd != INVALID_SOCKET)
 774         {
 775           WATCHPOINT_ASSERT(server->fd != INVALID_SOCKET);
 776           server->reset_socket();
 777         }
 778       }
 779     }
 780 #endif
 781     break;
 782
 783   case MEMCACHED_CONNECTION_UNIX_SOCKET:
 784     rc= unix_socket_connect(server);
 785     break;
 786   }
 787
 788   if (memcached_success(rc))
 789   {
 790     server->mark_server_as_clean();
 791     memcached_version_instance(server);
 792     return rc;
 793   }
 794   else if (set_last_disconnected)
 795   {
 796     set_last_disconnected_host(server);
 797     if (memcached_has_current_error(*server))
 798     {
 799       memcached_mark_server_for_timeout(server);
 800       assert(memcached_failed(memcached_instance_error_return(server)));
 801     }
 802     else
 803     {
 804       memcached_set_error(*server, rc, MEMCACHED_AT);
 805       memcached_mark_server_for_timeout(server);
 806     }
 807
 808     LIBMEMCACHED_MEMCACHED_CONNECT_END();
 809
 810     if (in_timeout)
 811     {
 812       char buffer[1024];
 813       int snprintf_length= snprintf(buffer, sizeof(buffer), "%s:%d", server->hostname(), int(server->port()));
 814       return memcached_set_error(*server, MEMCACHED_SERVER_TEMPORARILY_DISABLED, MEMCACHED_AT, buffer, snprintf_length);
 815     }
 816   }
 817
 818   return rc;
 819 }
 820
 821 memcached_return_t memcached_connect(memcached_instance_st* server)
 822 {
 823   return _memcached_connect(server, true);
 824 }