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