folly/io/async/AsyncServerSocket.cpp

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