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