git.m6w6.name Git - awesomized/libmemcached/blob - libmemcached/connect.c

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