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