folly/io/async/AsyncServerSocket.cpp

   1 /*
   2  * Copyright 2014 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #ifndef __STDC_FORMAT_MACROS
  18   #define __STDC_FORMAT_MACROS
  19 #endif
  20
  21 #include <folly/io/async/AsyncServerSocket.h>
  22
  23 #include <folly/io/async/EventBase.h>
  24 #include <folly/io/async/NotificationQueue.h>
  25 #include <folly/SocketAddress.h>
  26
  27 #include <errno.h>
  28 #include <string.h>
  29 #include <unistd.h>
  30 #include <fcntl.h>
  31 #include <sys/types.h>
  32 #include <sys/socket.h>
  33 #include <netinet/tcp.h>
  34
  35 namespace folly {
  36
  37 const uint32_t AsyncServerSocket::kDefaultMaxAcceptAtOnce;
  38 const uint32_t AsyncServerSocket::kDefaultCallbackAcceptAtOnce;
  39 const uint32_t AsyncServerSocket::kDefaultMaxMessagesInQueue;
  40
  41 int setCloseOnExec(int fd, int value) {
  42   // Read the current flags
  43   int old_flags = fcntl(fd, F_GETFD, 0);
  44
  45   // If reading the flags failed, return error indication now
  46   if (old_flags < 0)
  47     return -1;
  48
  49   // Set just the flag we want to set
  50   int new_flags;
  51   if (value != 0)
  52     new_flags = old_flags | FD_CLOEXEC;
  53   else
  54     new_flags = old_flags & ~FD_CLOEXEC;
  55
  56   // Store modified flag word in the descriptor
  57   return fcntl(fd, F_SETFD, new_flags);
  58 }
  59
  60 void AsyncServerSocket::RemoteAcceptor::start(
  61   EventBase* eventBase, uint32_t maxAtOnce, uint32_t maxInQueue) {
  62   setMaxReadAtOnce(maxAtOnce);
  63   queue_.setMaxQueueSize(maxInQueue);
  64
  65   if (!eventBase->runInEventBaseThread([=](){
  66         callback_->acceptStarted();
  67         this->startConsuming(eventBase, &queue_);
  68       })) {
  69     throw std::invalid_argument("unable to start waiting on accept "
  70                             "notification queue in the specified "
  71                             "EventBase thread");
  72   }
  73 }
  74
  75 void AsyncServerSocket::RemoteAcceptor::stop(
  76   EventBase* eventBase, AcceptCallback* callback) {
  77   if (!eventBase->runInEventBaseThread([=](){
  78         callback->acceptStopped();
  79         delete this;
  80       })) {
  81     throw std::invalid_argument("unable to start waiting on accept "
  82                             "notification queue in the specified "
  83                             "EventBase thread");
  84   }
  85 }
  86
  87 void AsyncServerSocket::RemoteAcceptor::messageAvailable(
  88   QueueMessage&& msg) {
  89
  90   switch (msg.type) {
  91     case MessageType::MSG_NEW_CONN:
  92     {
  93       callback_->connectionAccepted(msg.fd, msg.address);
  94       break;
  95     }
  96     case MessageType::MSG_ERROR:
  97     {
  98       std::runtime_error ex(msg.msg);
  99       callback_->acceptError(ex);
 100       break;
 101     }
 102     default:
 103     {
 104       LOG(ERROR) << "invalid accept notification message type "
 105                  << int(msg.type);
 106       std::runtime_error ex(
 107         "received invalid accept notification message type");
 108       callback_->acceptError(ex);
 109     }
 110   }
 111 }
 112
 113 /*
 114  * AsyncServerSocket::BackoffTimeout
 115  */
 116 class AsyncServerSocket::BackoffTimeout : public AsyncTimeout {
 117  public:
 118   BackoffTimeout(AsyncServerSocket* socket)
 119     : AsyncTimeout(socket->getEventBase()),
 120       socket_(socket) {}
 121
 122   virtual void timeoutExpired() noexcept {
 123     socket_->backoffTimeoutExpired();
 124   }
 125
 126  private:
 127   AsyncServerSocket* socket_;
 128 };
 129
 130 /*
 131  * AsyncServerSocket methods
 132  */
 133
 134 AsyncServerSocket::AsyncServerSocket(EventBase* eventBase)
 135 :   eventBase_(eventBase),
 136     accepting_(false),
 137     maxAcceptAtOnce_(kDefaultMaxAcceptAtOnce),
 138     maxNumMsgsInQueue_(kDefaultMaxMessagesInQueue),
 139     acceptRateAdjustSpeed_(0),
 140     acceptRate_(1),
 141     lastAccepTimestamp_(std::chrono::steady_clock::now()),
 142     numDroppedConnections_(0),
 143     callbackIndex_(0),
 144     backoffTimeout_(nullptr),
 145     callbacks_(),
 146     keepAliveEnabled_(true),
 147     closeOnExec_(true),
 148     shutdownSocketSet_(nullptr) {
 149 }
 150
 151 void AsyncServerSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
 152   if (shutdownSocketSet_ == newSS) {
 153     return;
 154   }
 155   if (shutdownSocketSet_) {
 156     for (auto& h : sockets_) {
 157       shutdownSocketSet_->remove(h.socket_);
 158     }
 159   }
 160   shutdownSocketSet_ = newSS;
 161   if (shutdownSocketSet_) {
 162     for (auto& h : sockets_) {
 163       shutdownSocketSet_->add(h.socket_);
 164     }
 165   }
 166 }
 167
 168 AsyncServerSocket::~AsyncServerSocket() {
 169   assert(callbacks_.empty());
 170 }
 171
 172 int AsyncServerSocket::stopAccepting(int shutdownFlags) {
 173   int result = 0;
 174   for (auto& handler : sockets_) {
 175     VLOG(10) << "AsyncServerSocket::stopAccepting " << this <<
 176               handler.socket_;
 177   }
 178   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 179
 180   // When destroy is called, unregister and close the socket immediately
 181   accepting_ = false;
 182
 183   for (auto& handler : sockets_) {
 184     handler.unregisterHandler();
 185     if (shutdownSocketSet_) {
 186       shutdownSocketSet_->close(handler.socket_);
 187     } else if (shutdownFlags >= 0) {
 188       result = ::shutdown(handler.socket_, shutdownFlags);
 189       pendingCloseSockets_.push_back(handler.socket_);
 190     } else {
 191       ::close(handler.socket_);
 192     }
 193   }
 194   sockets_.clear();
 195
 196   // Destroy the backoff timout.  This will cancel it if it is running.
 197   delete backoffTimeout_;
 198   backoffTimeout_ = nullptr;
 199
 200   // Close all of the callback queues to notify them that they are being
 201   // destroyed.  No one should access the AsyncServerSocket any more once
 202   // destroy() is called.  However, clear out callbacks_ before invoking the
 203   // accept callbacks just in case.  This will potentially help us detect the
 204   // bug if one of the callbacks calls addAcceptCallback() or
 205   // removeAcceptCallback().
 206   std::vector<CallbackInfo> callbacksCopy;
 207   callbacks_.swap(callbacksCopy);
 208   for (std::vector<CallbackInfo>::iterator it = callbacksCopy.begin();
 209        it != callbacksCopy.end();
 210        ++it) {
 211     it->consumer->stop(it->eventBase, it->callback);
 212   }
 213
 214   return result;
 215 }
 216
 217 void AsyncServerSocket::destroy() {
 218   stopAccepting();
 219   for (auto s: pendingCloseSockets_) {
 220     ::close(s);
 221   }
 222   // Then call DelayedDestruction::destroy() to take care of
 223   // whether or not we need immediate or delayed destruction
 224   DelayedDestruction::destroy();
 225 }
 226
 227 void AsyncServerSocket::attachEventBase(EventBase *eventBase) {
 228   assert(eventBase_ == nullptr);
 229   assert(eventBase->isInEventBaseThread());
 230
 231   eventBase_ = eventBase;
 232   for (auto& handler : sockets_) {
 233     handler.attachEventBase(eventBase);
 234   }
 235 }
 236
 237 void AsyncServerSocket::detachEventBase() {
 238   assert(eventBase_ != nullptr);
 239   assert(eventBase_->isInEventBaseThread());
 240   assert(!accepting_);
 241
 242   eventBase_ = nullptr;
 243   for (auto& handler : sockets_) {
 244     handler.detachEventBase();
 245   }
 246 }
 247
 248 void AsyncServerSocket::useExistingSockets(const std::vector<int>& fds) {
 249   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 250
 251   if (sockets_.size() > 0) {
 252     throw std::invalid_argument(
 253                               "cannot call useExistingSocket() on a "
 254                               "AsyncServerSocket that already has a socket");
 255   }
 256
 257   for (auto fd: fds) {
 258     // Set addressFamily_ from this socket.
 259     // Note that the socket may not have been bound yet, but
 260     // setFromLocalAddress() will still work and get the correct address family.
 261     // We will update addressFamily_ again anyway if bind() is called later.
 262     SocketAddress address;
 263     address.setFromLocalAddress(fd);
 264
 265     setupSocket(fd);
 266     sockets_.push_back(
 267       ServerEventHandler(eventBase_, fd, this, address.getFamily()));
 268     sockets_.back().changeHandlerFD(fd);
 269   }
 270 }
 271
 272 void AsyncServerSocket::useExistingSocket(int fd) {
 273   useExistingSockets({fd});
 274 }
 275
 276 void AsyncServerSocket::bindSocket(
 277     int fd,
 278     const SocketAddress& address,
 279     bool isExistingSocket) {
 280   sockaddr_storage addrStorage;
 281   address.getAddress(&addrStorage);
 282   sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 283   if (::bind(fd, saddr, address.getActualSize()) != 0) {
 284     if (!isExistingSocket) {
 285       ::close(fd);
 286     }
 287     folly::throwSystemError(errno,
 288         "failed to bind to async server socket: " +
 289         address.describe());
 290   }
 291
 292   // If we just created this socket, update the EventHandler and set socket_
 293   if (!isExistingSocket) {
 294     sockets_.push_back(
 295       ServerEventHandler(eventBase_, fd, this, address.getFamily()));
 296   }
 297 }
 298
 299 void AsyncServerSocket::bind(const SocketAddress& address) {
 300   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 301
 302   // useExistingSocket() may have been called to initialize socket_ already.
 303   // However, in the normal case we need to create a new socket now.
 304   // Don't set socket_ yet, so that socket_ will remain uninitialized if an
 305   // error occurs.
 306   int fd;
 307   if (sockets_.size() == 0) {
 308     fd = createSocket(address.getFamily());
 309   } else if (sockets_.size() == 1) {
 310     if (address.getFamily() != sockets_[0].addressFamily_) {
 311       throw std::invalid_argument(
 312                                 "Attempted to bind address to socket with "
 313                                 "different address family");
 314     }
 315     fd = sockets_[0].socket_;
 316   } else {
 317     throw std::invalid_argument(
 318                               "Attempted to bind to multiple fds");
 319   }
 320
 321   bindSocket(fd, address, !sockets_.empty());
 322 }
 323
 324 void AsyncServerSocket::bind(
 325     const std::vector<IPAddress>& ipAddresses,
 326     uint16_t port) {
 327   if (ipAddresses.empty()) {
 328     throw std::invalid_argument("No ip addresses were provided");
 329   }
 330   if (!sockets_.empty()) {
 331     throw std::invalid_argument("Cannot call bind on a AsyncServerSocket "
 332                                 "that already has a socket.");
 333   }
 334
 335   for (const IPAddress& ipAddress : ipAddresses) {
 336     SocketAddress address(ipAddress.toFullyQualified(), port);
 337     int fd = createSocket(address.getFamily());
 338
 339     bindSocket(fd, address, false);
 340   }
 341   if (sockets_.size() == 0) {
 342     throw std::runtime_error(
 343         "did not bind any async server socket for port and addresses");
 344   }
 345 }
 346
 347 void AsyncServerSocket::bind(uint16_t port) {
 348   struct addrinfo hints, *res, *res0;
 349   char sport[sizeof("65536")];
 350
 351   memset(&hints, 0, sizeof(hints));
 352   hints.ai_family = AF_UNSPEC;
 353   hints.ai_socktype = SOCK_STREAM;
 354   hints.ai_flags = AI_PASSIVE;
 355   snprintf(sport, sizeof(sport), "%u", port);
 356
 357   if (getaddrinfo(nullptr, sport, &hints, &res0)) {
 358     throw std::invalid_argument(
 359                               "Attempted to bind address to socket with "
 360                               "bad getaddrinfo");
 361   }
 362
 363   folly::ScopeGuard guard = folly::makeGuard([&]{
 364       freeaddrinfo(res0);
 365     });
 366   DCHECK(&guard);
 367
 368   auto setupAddress = [&] (struct addrinfo* res) {
 369     int s = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
 370     // IPv6/IPv4 may not be supported by the kernel
 371     if (s < 0 && errno == EAFNOSUPPORT) {
 372       return;
 373     }
 374     CHECK_GE(s, 0);
 375
 376     try {
 377       setupSocket(s);
 378     } catch (...) {
 379       ::close(s);
 380       throw;
 381     }
 382
 383     if (res->ai_family == AF_INET6) {
 384       int v6only = 1;
 385       CHECK(0 == setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY,
 386                             &v6only, sizeof(v6only)));
 387     }
 388
 389     SocketAddress address;
 390     address.setFromLocalAddress(s);
 391
 392     sockets_.push_back(
 393       ServerEventHandler(eventBase_, s, this, address.getFamily()));
 394
 395     // Bind to the socket
 396     if (::bind(s, res->ai_addr, res->ai_addrlen) != 0) {
 397       folly::throwSystemError(
 398         errno,
 399         "failed to bind to async server socket for port");
 400     }
 401   };
 402
 403   // Prefer AF_INET6 addresses. RFC 3484 mandates that getaddrinfo
 404   // should return IPv6 first and then IPv4 addresses, but glibc's
 405   // getaddrinfo(nullptr) with AI_PASSIVE returns:
 406   // - 0.0.0.0 (IPv4-only)
 407   // - :: (IPv6+IPv4) in this order
 408   // See: https://sourceware.org/bugzilla/show_bug.cgi?id=9981
 409   for (res = res0; res; res = res->ai_next) {
 410     if (res->ai_family == AF_INET6) {
 411       setupAddress(res);
 412     }
 413   }
 414
 415   for (res = res0; res; res = res->ai_next) {
 416     if (res->ai_family != AF_INET6) {
 417       setupAddress(res);
 418     }
 419   }
 420
 421   if (sockets_.size() == 0) {
 422     throw std::runtime_error(
 423         "did not bind any async server socket for port");
 424   }
 425 }
 426
 427 void AsyncServerSocket::listen(int backlog) {
 428   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 429
 430   // Start listening
 431   for (auto& handler : sockets_) {
 432     if (::listen(handler.socket_, backlog) == -1) {
 433       folly::throwSystemError(errno,
 434                                     "failed to listen on async server socket");
 435     }
 436   }
 437 }
 438
 439 void AsyncServerSocket::getAddress(SocketAddress* addressReturn) const {
 440   CHECK(sockets_.size() >= 1);
 441   VLOG_IF(2, sockets_.size() > 1)
 442     << "Warning: getAddress() called and multiple addresses available ("
 443     << sockets_.size() << "). Returning only the first one.";
 444
 445   addressReturn->setFromLocalAddress(sockets_[0].socket_);
 446 }
 447
 448 std::vector<SocketAddress> AsyncServerSocket::getAddresses()
 449     const {
 450   CHECK(sockets_.size() >= 1);
 451   auto tsaVec = std::vector<SocketAddress>(sockets_.size());
 452   auto tsaIter = tsaVec.begin();
 453   for (const auto& socket : sockets_) {
 454     (tsaIter++)->setFromLocalAddress(socket.socket_);
 455   };
 456   return tsaVec;
 457 }
 458
 459 void AsyncServerSocket::addAcceptCallback(AcceptCallback *callback,
 460                                            EventBase *eventBase,
 461                                            uint32_t maxAtOnce) {
 462   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 463
 464   // If this is the first accept callback and we are supposed to be accepting,
 465   // start accepting once the callback is installed.
 466   bool runStartAccepting = accepting_ && callbacks_.empty();
 467
 468   if (!eventBase) {
 469     eventBase = eventBase_; // Run in AsyncServerSocket's eventbase
 470   }
 471
 472   callbacks_.push_back(CallbackInfo(callback, eventBase));
 473
 474   // Start the remote acceptor.
 475   //
 476   // It would be nice if we could avoid starting the remote acceptor if
 477   // eventBase == eventBase_.  However, that would cause issues if
 478   // detachEventBase() and attachEventBase() were ever used to change the
 479   // primary EventBase for the server socket.  Therefore we require the caller
 480   // to specify a nullptr EventBase if they want to ensure that the callback is
 481   // always invoked in the primary EventBase, and to be able to invoke that
 482   // callback more efficiently without having to use a notification queue.
 483   RemoteAcceptor* acceptor = nullptr;
 484   try {
 485     acceptor = new RemoteAcceptor(callback);
 486     acceptor->start(eventBase, maxAtOnce, maxNumMsgsInQueue_);
 487   } catch (...) {
 488     callbacks_.pop_back();
 489     delete acceptor;
 490     throw;
 491   }
 492   callbacks_.back().consumer = acceptor;
 493
 494   // If this is the first accept callback and we are supposed to be accepting,
 495   // start accepting.
 496   if (runStartAccepting) {
 497     startAccepting();
 498   }
 499 }
 500
 501 void AsyncServerSocket::removeAcceptCallback(AcceptCallback *callback,
 502                                               EventBase *eventBase) {
 503   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 504
 505   // Find the matching AcceptCallback.
 506   // We just do a simple linear search; we don't expect removeAcceptCallback()
 507   // to be called frequently, and we expect there to only be a small number of
 508   // callbacks anyway.
 509   std::vector<CallbackInfo>::iterator it = callbacks_.begin();
 510   uint32_t n = 0;
 511   while (true) {
 512     if (it == callbacks_.end()) {
 513       throw std::runtime_error("AsyncServerSocket::removeAcceptCallback(): "
 514                               "accept callback not found");
 515     }
 516     if (it->callback == callback &&
 517         (it->eventBase == eventBase || eventBase == nullptr)) {
 518       break;
 519     }
 520     ++it;
 521     ++n;
 522   }
 523
 524   // Remove this callback from callbacks_.
 525   //
 526   // Do this before invoking the acceptStopped() callback, in case
 527   // acceptStopped() invokes one of our methods that examines callbacks_.
 528   //
 529   // Save a copy of the CallbackInfo first.
 530   CallbackInfo info(*it);
 531   callbacks_.erase(it);
 532   if (n < callbackIndex_) {
 533     // We removed an element before callbackIndex_.  Move callbackIndex_ back
 534     // one step, since things after n have been shifted back by 1.
 535     --callbackIndex_;
 536   } else {
 537     // We removed something at or after callbackIndex_.
 538     // If we removed the last element and callbackIndex_ was pointing at it,
 539     // we need to reset callbackIndex_ to 0.
 540     if (callbackIndex_ >= callbacks_.size()) {
 541       callbackIndex_ = 0;
 542     }
 543   }
 544
 545   info.consumer->stop(info.eventBase, info.callback);
 546
 547   // If we are supposed to be accepting but the last accept callback
 548   // was removed, unregister for events until a callback is added.
 549   if (accepting_ && callbacks_.empty()) {
 550     for (auto& handler : sockets_) {
 551       handler.unregisterHandler();
 552     }
 553   }
 554 }
 555
 556 void AsyncServerSocket::startAccepting() {
 557   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 558
 559   accepting_ = true;
 560   if (callbacks_.empty()) {
 561     // We can't actually begin accepting if no callbacks are defined.
 562     // Wait until a callback is added to start accepting.
 563     return;
 564   }
 565
 566   for (auto& handler : sockets_) {
 567     if (!handler.registerHandler(
 568           EventHandler::READ | EventHandler::PERSIST)) {
 569       throw std::runtime_error("failed to register for accept events");
 570     }
 571   }
 572 }
 573
 574 void AsyncServerSocket::pauseAccepting() {
 575   assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
 576   accepting_ = false;
 577   for (auto& handler : sockets_) {
 578    handler. unregisterHandler();
 579   }
 580
 581   // If we were in the accept backoff state, disable the backoff timeout
 582   if (backoffTimeout_) {
 583     backoffTimeout_->cancelTimeout();
 584   }
 585 }
 586
 587 int AsyncServerSocket::createSocket(int family) {
 588   int fd = socket(family, SOCK_STREAM, 0);
 589   if (fd == -1) {
 590     folly::throwSystemError(errno, "error creating async server socket");
 591   }
 592
 593   try {
 594     setupSocket(fd);
 595   } catch (...) {
 596     ::close(fd);
 597     throw;
 598   }
 599   return fd;
 600 }
 601
 602 void AsyncServerSocket::setupSocket(int fd) {
 603   // Get the address family
 604   SocketAddress address;
 605   address.setFromLocalAddress(fd);
 606   auto family = address.getFamily();
 607
 608   // Put the socket in non-blocking mode
 609   if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
 610     folly::throwSystemError(errno,
 611                             "failed to put socket in non-blocking mode");
 612   }
 613
 614   // Set reuseaddr to avoid 2MSL delay on server restart
 615   int one = 1;
 616   if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one)) != 0) {
 617     // This isn't a fatal error; just log an error message and continue
 618     LOG(ERROR) << "failed to set SO_REUSEADDR on async server socket " << errno;
 619   }
 620
 621   // Set reuseport to support multiple accept threads
 622   int zero = 0;
 623   if (reusePortEnabled_ &&
 624       setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(int)) != 0) {
 625     LOG(ERROR) << "failed to set SO_REUSEPORT on async server socket "
 626                << strerror(errno);
 627     folly::throwSystemError(errno,
 628                             "failed to bind to async server socket: " +
 629                             address.describe());
 630   }
 631
 632   // Set keepalive as desired
 633   if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE,
 634                  (keepAliveEnabled_) ? &one : &zero, sizeof(int)) != 0) {
 635     LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: " <<
 636             strerror(errno);
 637   }
 638
 639   // Setup FD_CLOEXEC flag
 640   if (closeOnExec_ &&
 641       (-1 == folly::setCloseOnExec(fd, closeOnExec_))) {
 642     LOG(ERROR) << "failed to set FD_CLOEXEC on async server socket: " <<
 643             strerror(errno);
 644   }
 645
 646   // Set TCP nodelay if available, MAC OS X Hack
 647   // See http://lists.danga.com/pipermail/memcached/2005-March/001240.html
 648 #ifndef TCP_NOPUSH
 649   if (family != AF_UNIX) {
 650     if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one)) != 0) {
 651       // This isn't a fatal error; just log an error message and continue
 652       LOG(ERROR) << "failed to set TCP_NODELAY on async server socket: " <<
 653               strerror(errno);
 654     }
 655   }
 656 #endif
 657
 658   if (shutdownSocketSet_) {
 659     shutdownSocketSet_->add(fd);
 660   }
 661 }
 662
 663 void AsyncServerSocket::handlerReady(
 664   uint16_t events, int fd, sa_family_t addressFamily) noexcept {
 665   assert(!callbacks_.empty());
 666   DestructorGuard dg(this);
 667
 668   // Only accept up to maxAcceptAtOnce_ connections at a time,
 669   // to avoid starving other I/O handlers using this EventBase.
 670   for (uint32_t n = 0; n < maxAcceptAtOnce_; ++n) {
 671     SocketAddress address;
 672
 673     sockaddr_storage addrStorage;
 674     socklen_t addrLen = sizeof(addrStorage);
 675     sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
 676
 677     // In some cases, accept() doesn't seem to update these correctly.
 678     saddr->sa_family = addressFamily;
 679     if (addressFamily == AF_UNIX) {
 680       addrLen = sizeof(struct sockaddr_un);
 681     }
 682
 683     // Accept a new client socket
 684 #ifdef SOCK_NONBLOCK
 685     int clientSocket = accept4(fd, saddr, &addrLen, SOCK_NONBLOCK);
 686 #else
 687     int clientSocket = accept(fd, saddr, &addrLen);
 688 #endif
 689
 690     address.setFromSockaddr(saddr, addrLen);
 691
 692     std::chrono::time_point<std::chrono::steady_clock> nowMs =
 693       std::chrono::steady_clock::now();
 694     int64_t timeSinceLastAccept = std::max(
 695       int64_t(0),
 696       nowMs.time_since_epoch().count() -
 697       lastAccepTimestamp_.time_since_epoch().count());
 698     lastAccepTimestamp_ = nowMs;
 699     if (acceptRate_ < 1) {
 700       acceptRate_ *= 1 + acceptRateAdjustSpeed_ * timeSinceLastAccept;
 701       if (acceptRate_ >= 1) {
 702         acceptRate_ = 1;
 703       } else if (rand() > acceptRate_ * RAND_MAX) {
 704         ++numDroppedConnections_;
 705         if (clientSocket >= 0) {
 706           ::close(clientSocket);
 707         }
 708         continue;
 709       }
 710     }
 711
 712     if (clientSocket < 0) {
 713       if (errno == EAGAIN) {
 714         // No more sockets to accept right now.
 715         // Check for this code first, since it's the most common.
 716         return;
 717       } else if (errno == EMFILE || errno == ENFILE) {
 718         // We're out of file descriptors.  Perhaps we're accepting connections
 719         // too quickly. Pause accepting briefly to back off and give the server
 720         // a chance to recover.
 721         LOG(ERROR) << "accept failed: out of file descriptors; entering accept "
 722                 "back-off state";
 723         enterBackoff();
 724
 725         // Dispatch the error message
 726         dispatchError("accept() failed", errno);
 727       } else {
 728         dispatchError("accept() failed", errno);
 729       }
 730       return;
 731     }
 732
 733 #ifndef SOCK_NONBLOCK
 734     // Explicitly set the new connection to non-blocking mode
 735     if (fcntl(clientSocket, F_SETFL, O_NONBLOCK) != 0) {
 736       ::close(clientSocket);
 737       dispatchError("failed to set accepted socket to non-blocking mode",
 738                     errno);
 739       return;
 740     }
 741 #endif
 742
 743     // Inform the callback about the new connection
 744     dispatchSocket(clientSocket, std::move(address));
 745
 746     // If we aren't accepting any more, break out of the loop
 747     if (!accepting_ || callbacks_.empty()) {
 748       break;
 749     }
 750   }
 751 }
 752
 753 void AsyncServerSocket::dispatchSocket(int socket,
 754                                         SocketAddress&& address) {
 755   uint32_t startingIndex = callbackIndex_;
 756
 757   // Short circuit if the callback is in the primary EventBase thread
 758
 759   CallbackInfo *info = nextCallback();
 760   if (info->eventBase == nullptr) {
 761     info->callback->connectionAccepted(socket, address);
 762     return;
 763   }
 764
 765   // Create a message to send over the notification queue
 766   QueueMessage msg;
 767   msg.type = MessageType::MSG_NEW_CONN;
 768   msg.address = std::move(address);
 769   msg.fd = socket;
 770
 771   // Loop until we find a free queue to write to
 772   while (true) {
 773     if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
 774       // Success! return.
 775       return;
 776    }
 777
 778     // We couldn't add to queue.  Fall through to below
 779
 780     ++numDroppedConnections_;
 781     if (acceptRateAdjustSpeed_ > 0) {
 782       // aggressively decrease accept rate when in trouble
 783       static const double kAcceptRateDecreaseSpeed = 0.1;
 784       acceptRate_ *= 1 - kAcceptRateDecreaseSpeed;
 785     }
 786
 787
 788     if (callbackIndex_ == startingIndex) {
 789       // The notification queue was full
 790       // We can't really do anything at this point other than close the socket.
 791       //
 792       // This should only happen if a user's service is behaving extremely
 793       // badly and none of the EventBase threads are looping fast enough to
 794       // process the incoming connections.  If the service is overloaded, it
 795       // should use pauseAccepting() to temporarily back off accepting new
 796       // connections, before they reach the point where their threads can't
 797       // even accept new messages.
 798       LOG(ERROR) << "failed to dispatch newly accepted socket:"
 799                  << " all accept callback queues are full";
 800       ::close(socket);
 801       return;
 802     }
 803
 804     info = nextCallback();
 805   }
 806 }
 807
 808 void AsyncServerSocket::dispatchError(const char *msgstr, int errnoValue) {
 809   uint32_t startingIndex = callbackIndex_;
 810   CallbackInfo *info = nextCallback();
 811
 812   // Create a message to send over the notification queue
 813   QueueMessage msg;
 814   msg.type = MessageType::MSG_ERROR;
 815   msg.err = errnoValue;
 816   msg.msg = std::move(msgstr);
 817
 818   while (true) {
 819     // Short circuit if the callback is in the primary EventBase thread
 820     if (info->eventBase == nullptr) {
 821       std::runtime_error ex(
 822         std::string(msgstr) +  folly::to<std::string>(errnoValue));
 823       info->callback->acceptError(ex);
 824       return;
 825     }
 826
 827     if (info->consumer->getQueue()->tryPutMessageNoThrow(std::move(msg))) {
 828       return;
 829     }
 830     // Fall through and try another callback
 831
 832     if (callbackIndex_ == startingIndex) {
 833       // The notification queues for all of the callbacks were full.
 834       // We can't really do anything at this point.
 835       LOG(ERROR) << "failed to dispatch accept error: all accept callback "
 836         "queues are full: error msg:  " <<
 837         msg.msg.c_str() << errnoValue;
 838       return;
 839     }
 840     info = nextCallback();
 841   }
 842 }
 843
 844 void AsyncServerSocket::enterBackoff() {
 845   // If this is the first time we have entered the backoff state,
 846   // allocate backoffTimeout_.
 847   if (backoffTimeout_ == nullptr) {
 848     try {
 849       backoffTimeout_ = new BackoffTimeout(this);
 850     } catch (const std::bad_alloc& ex) {
 851       // Man, we couldn't even allocate the timer to re-enable accepts.
 852       // We must be in pretty bad shape.  Don't pause accepting for now,
 853       // since we won't be able to re-enable ourselves later.
 854       LOG(ERROR) << "failed to allocate AsyncServerSocket backoff"
 855                  << " timer; unable to temporarly pause accepting";
 856       return;
 857     }
 858   }
 859
 860   // For now, we simply pause accepting for 1 second.
 861   //
 862   // We could add some smarter backoff calculation here in the future.  (e.g.,
 863   // start sleeping for longer if we keep hitting the backoff frequently.)
 864   // Typically the user needs to figure out why the server is overloaded and
 865   // fix it in some other way, though.  The backoff timer is just a simple
 866   // mechanism to try and give the connection processing code a little bit of
 867   // breathing room to catch up, and to avoid just spinning and failing to
 868   // accept over and over again.
 869   const uint32_t timeoutMS = 1000;
 870   if (!backoffTimeout_->scheduleTimeout(timeoutMS)) {
 871     LOG(ERROR) << "failed to schedule AsyncServerSocket backoff timer;"
 872                << "unable to temporarly pause accepting";
 873     return;
 874   }
 875
 876   // The backoff timer is scheduled to re-enable accepts.
 877   // Go ahead and disable accepts for now.  We leave accepting_ set to true,
 878   // since that tracks the desired state requested by the user.
 879   for (auto& handler : sockets_) {
 880     handler.unregisterHandler();
 881   }
 882 }
 883
 884 void AsyncServerSocket::backoffTimeoutExpired() {
 885   // accepting_ should still be true.
 886   // If pauseAccepting() was called while in the backoff state it will cancel
 887   // the backoff timeout.
 888   assert(accepting_);
 889   // We can't be detached from the EventBase without being paused
 890   assert(eventBase_ != nullptr && eventBase_->isInEventBaseThread());
 891
 892   // If all of the callbacks were removed, we shouldn't re-enable accepts
 893   if (callbacks_.empty()) {
 894     return;
 895   }
 896
 897   // Register the handler.
 898   for (auto& handler : sockets_) {
 899     if (!handler.registerHandler(
 900           EventHandler::READ | EventHandler::PERSIST)) {
 901       // We're hosed.  We could just re-schedule backoffTimeout_ to
 902       // re-try again after a little bit.  However, we don't want to
 903       // loop retrying forever if we can't re-enable accepts.  Just
 904       // abort the entire program in this state; things are really bad
 905       // and restarting the entire server is probably the best remedy.
 906       LOG(ERROR)
 907         << "failed to re-enable AsyncServerSocket accepts after backoff; "
 908         << "crashing now";
 909       abort();
 910     }
 911   }
 912 }
 913
 914
 915
 916 } // folly