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