folly/io/async/AsyncServerSocket.cpp

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