folly/io/async/AsyncServerSocket.cpp

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