folly/io/async/AsyncServerSocket.cpp

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